Banco vetorial na prática: pgvector e além

O problema real

Você leu que precisa de um "banco vetorial" e foi procurar qual escolher: Pinecone, Weaviate, Milvus, Qdrant. Assinou Pinecone (USD 0,04 por 100k embeddings armazenados/mês), criou uma infra nova, aumentou a complexidade da stack. Meses depois, descobre que o Postgres que já tem em produção resolveria 95% do seu caso com pgvector, economizando custo e devops. O 1% que não resolve é bem específico — escala massiva (>100M embeddings) ou filtros muito complexos.

A verdade: você não precisa de um banco dedicado até precisar. Começar com pgvector é pragmático, e migrar depois é trivial.

Por que pgvector cobre a maioria dos casos?

Um "banco vetorial" em sua essência é:

Armazenar vetores (embeddings) de forma indexada
Buscar rápido por similaridade (cosseno, euclidiana, ou inner product)
Filtrar por metadados enquanto busca

Postgres + pgvector faz exatamente isso. Índices HNSW mantêm buscas em O(log n). Para 1 milhão de embeddings em sa-east-1, latência tipicamente entre 50-200ms. Custo: zero (você já paga Postgres). Migração futura: SQL direto em Python, sem aprender SDK novo.

Mão na massa

Vamos montar um retriever de RAG real usando pgvector com SQL.

Setup: instalar pgvector

# Em um Postgres existente
docker exec <seu_container_postgres> psql -U postgres

# Dentro do psql:
CREATE EXTENSION IF NOT EXISTS vector;

# Verifica instalação
SELECT * FROM pg_extension WHERE extname = 'vector';

Criar tabela com coluna vetorial

-- Cria tabela para armazenar chunks + embeddings
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),  -- 1536 dimensões é padrão OpenAI
    metadata JSONB,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

-- Índice para buscas rápidas (HNSW é mais rápido que IVFFlat para <5M vetores)
CREATE INDEX ON document_chunks 
    USING hnsw (embedding vector_cosine_ops)
    WITH (m = 16, ef_construction = 64);

-- Índice para filtros de metadados
CREATE INDEX ON document_chunks USING GIN (metadata);

Inserir embeddings

import psycopg2
from openai import OpenAI
from typing import List

def insert_chunks_with_embeddings(
    db_host: str,
    db_name: str,
    document_id: int,
    chunks: List[str],
    embedding_model: str = "text-embedding-3-small"
):
    """Insere chunks com embeddings no pgvector."""
    client = OpenAI()
    
    # Gera embeddings em batch (até 2.048 por request)
    response = client.embeddings.create(
        model=embedding_model,
        input=chunks
    )
    embeddings = [item.embedding for item in response.data]
    
    # Conecta ao Postgres
    conn = psycopg2.connect(
        host=db_host,
        database=db_name,
        user="postgres",
        password="sua_senha"
    )
    cursor = conn.cursor()
    
    # Insere chunks + embeddings
    for idx, (chunk, embedding) in enumerate(zip(chunks, embeddings)):
        cursor.execute(
            """
            INSERT INTO document_chunks 
            (document_id, chunk_index, content, embedding, metadata)
            VALUES (%s, %s, %s, %s, %s)
            """,
            (
                document_id,
                idx,
                chunk,
                embedding,  # pgvector aceita list direto
                {"source": "documento_principal"}
            )
        )
    
    conn.commit()
    cursor.close()
    conn.close()
    
    print(f"Inseridos {len(chunks)} chunks com embeddings")

# Uso
chunks = [
    "A Constituição foi promulgada em 1988",
    "O Presidente é eleito por voto direto",
    "O Congresso Nacional composto de Câmara e Senado"
]

insert_chunks_with_embeddings(
    db_host="localhost",
    db_name="sua_db",
    document_id=1,
    chunks=chunks
)

Custo real: Para 10 mil chunks de 512 tokens: 5M tokens × USD 0,02/1M = USD 0,10. Uma vez. (Depois é grátis — apenas Postgres.)

Busca por similaridade (cosseno)

def retrieve_by_similarity(
    db_host: str,
    db_name: str,
    query: str,
    top_k: int = 3,
    embedding_model: str = "text-embedding-3-small"
) -> List[dict]:
    """Busca chunks similares à query usando cosseno."""
    
    client = OpenAI()
    
    # Embedding da query
    response = client.embeddings.create(
        model=embedding_model,
        input=[query]
    )
    query_embedding = response.data[0].embedding
    
    # Busca no Postgres
    conn = psycopg2.connect(
        host=db_host,
        database=db_name,
        user="postgres",
        password="sua_senha"
    )
    cursor = conn.cursor()
    
    # HNSW index (mais rápido com <5M vetores)
    cursor.execute(
        """
        SELECT 
            id,
            content,
            metadata,
            1 - (embedding <=> %s) as similarity_score
        FROM document_chunks
        ORDER BY embedding <=> %s
        LIMIT %s
        """,
        (query_embedding, query_embedding, top_k)
    )
    
    results = cursor.fetchall()
    cursor.close()
    conn.close()
    
    return [
        {
            "id": r[0],
            "content": r[1],
            "metadata": r[2],
            "score": r[3]
        }
        for r in results
    ]

# Uso
query = "Qual é o sistema de governo?"
results = retrieve_by_similarity(
    db_host="localhost",
    db_name="sua_db",
    query=query,
    top_k=3
)

for r in results:
    print(f"Score: {r['score']:.3f} | {r['content'][:80]}...")

Latência típica: 50-150ms para 1M documentos em sa-east-1.

Busca híbrida (vetorial + full-text)

Combine busca semântica (embedding) com busca literal (full-text). Melhor de dois mundos.

def retrieve_hybrid(
    db_host: str,
    db_name: str,
    query: str,
    top_k: int = 3
) -> List[dict]:
    """Busca híbrida: 70% semântica + 30% full-text."""
    
    client = OpenAI()
    
    # Embedding da query
    response = client.embeddings.create(
        model="text-embedding-3-small",
        input=[query]
    )
    query_embedding = response.data[0].embedding
    
    conn = psycopg2.connect(
        host=db_host,
        database=db_name,
        user="postgres",
        password="sua_senha"
    )
    cursor = conn.cursor()
    
    # CTE com ranking: 70% similaridade vetorial + 30% full-text
    cursor.execute(
        """
        WITH vector_results AS (
            SELECT 
                id,
                content,
                metadata,
                1 - (embedding <=> %s) as vector_score,
                ROW_NUMBER() OVER (ORDER BY embedding <=> %s) as vector_rank
            FROM document_chunks
            LIMIT 100  -- busca os top-100 primeiro
        ),
        text_results AS (
            SELECT 
                id,
                content,
                metadata,
                ts_rank(to_tsvector('portuguese', content), 
                        plainto_tsquery('portuguese', %s)) as text_score,
                ROW_NUMBER() OVER (ORDER BY ts_rank DESC) as text_rank
            FROM document_chunks
            WHERE to_tsvector('portuguese', content) @@ 
                  plainto_tsquery('portuguese', %s)
        ),
        combined AS (
            SELECT 
                COALESCE(v.id, t.id) as id,
                COALESCE(v.content, t.content) as content,
                COALESCE(v.metadata, t.metadata) as metadata,
                0.7 * COALESCE(v.vector_score, 0) + 
                0.3 * COALESCE(t.text_score / 32.0, 0) as combined_score
            FROM vector_results v
            FULL OUTER JOIN text_results t ON v.id = t.id
        )
        SELECT id, content, metadata, combined_score
        FROM combined
        ORDER BY combined_score DESC
        LIMIT %s
        """,
        (query_embedding, query_embedding, query, query, top_k)
    )
    
    results = cursor.fetchall()
    cursor.close()
    conn.close()
    
    return [
        {
            "id": r[0],
            "content": r[1],
            "metadata": r[2],
            "score": r[3]
        }
        for r in results
    ]

Vantagem: Captura tanto semântica ("qual sistema de governo") quanto literal ("constituição", "congresso"). Melhora recall.

Filtros de metadados durante busca

def retrieve_with_filters(
    db_host: str,
    db_name: str,
    query: str,
    filters: dict,
    top_k: int = 3
) -> List[dict]:
    """Busca vetorial + filtros de metadados (ex: apenas documentos após 2020)."""
    
    client = OpenAI()
    response = client.embeddings.create(
        model="text-embedding-3-small",
        input=[query]
    )
    query_embedding = response.data[0].embedding
    
    conn = psycopg2.connect(
        host=db_host,
        database=db_name,
        user="postgres",
        password="sua_senha"
    )
    cursor = conn.cursor()
    
    # Constrói WHERE clause dinamicamente
    where_clauses = ["1=1"]
    params = [query_embedding, query_embedding]
    
    for key, value in filters.items():
        where_clauses.append(f"metadata->'{key}' = %s")
        params.append(value)
    
    where_sql = " AND ".join(where_clauses)
    
    cursor.execute(
        f"""
        SELECT 
            id,
            content,
            metadata,
            1 - (embedding <=> %s) as similarity_score
        FROM document_chunks
        WHERE {where_sql}
        ORDER BY embedding <=> %s
        LIMIT %s
        """,
        params + [top_k]
    )
    
    results = cursor.fetchall()
    cursor.close()
    conn.close()
    
    return results

# Uso: buscar só documentos da legislação pós-2020
results = retrieve_with_filters(
    db_host="localhost",
    db_name="sua_db",
    query="governo",
    filters={"ano": "2020"},
    top_k=5
)

Armadilhas comuns

Armadilha 1: Índice HNSW vs IVFFlat

HNSW: Rápido, recall alto (~99%), mas construção lenta (5-10 min para 1M). Use para < 5M vetores.
IVFFlat: Construção rápida, mas recall ~95% e requer re-tuning. Use para > 5M vetores.

Para começar, use HNSW. Se latência ficar insuportável, mude.

Armadilha 2: Não normalizar embeddings

Se misturar modelos de embedding (OpenAI com Hugging Face), distâncias ficam inconsistentes. Use sempre o mesmo modelo.

Armadilha 3: Forget to vacuum

Postgres não deleta automaticamente. Se deletar muitos chunks:

VACUUM ANALYZE document_chunks;

Quando não usar pgvector

Escala > 100M embeddings: Postgres fica lento. Use Pinecone (USD 40/mês) ou Weaviate.
Filtros muito complexos: Você quer buscar "documentos de 2023 para usuários premium em sa-east-1 com score > 0.8". Weaviate é melhor.
Multi-tenancy com isolamento rígido: Postgres consegue, mas é complexo. Pinecone/Weaviate isolam melhor.
Zero latência é crítico: Caching em Redis com pgvector bate tudo, mas adiciona complexidade.

Próximos passos

Combine pgvector com RAG na prática — substitua Chroma por pgvector.
Leia FinOps para IA para calcular real vs. Pinecone em escala.
Se evoluir para multi-tenant, migre com observabilidade de LLM em place.

Resumo prático

Aspecto	pgvector	Pinecone	Weaviate
Setup	5 min	2 min	30 min
Custo (1M docs)	Postgres	USD 40/mês	USD 50/mês+
Latência (sa-east-1)	50-150ms	10-50ms	50-200ms
Escala máxima	100M (degradado)	∞	500M (caro)
Filtros complexos	⭐⭐⭐	⭐⭐	⭐⭐⭐⭐⭐

Recomendação: Comece com pgvector. Se escala > 10M ou latência < 10ms virar crítico, migre. SQL em Postgres é transferível — seus evals vão continuar funcionando.

Banco vetorial na prática: pgvector e além

O problema real

Por que pgvector cobre a maioria dos casos?

Mão na massa

Setup: instalar pgvector

Criar tabela com coluna vetorial

Inserir embeddings

Busca por similaridade (cosseno)

Busca híbrida (vetorial + full-text)

Filtros de metadados durante busca

Armadilhas comuns

Armadilha 1: Índice HNSW vs IVFFlat

Armadilha 2: Não normalizar embeddings

Armadilha 3: Forget to vacuum

Quando não usar pgvector

Próximos passos

Resumo prático

Continue lendo

Chunking: as estratégias que funcionam

RAG na prática: do zero ao primeiro retriever

RAG, fine-tuning ou prompt? Guia de decisão

O problema real#

Por que pgvector cobre a maioria dos casos?#

Mão na massa#

Setup: instalar pgvector#

Criar tabela com coluna vetorial#

Inserir embeddings#

Busca por similaridade (cosseno)#

Busca híbrida (vetorial + full-text)#

Filtros de metadados durante busca#

Armadilhas comuns#

Armadilha 1: Índice HNSW vs IVFFlat#

Armadilha 2: Não normalizar embeddings#

Armadilha 3: Forget to vacuum#

Quando não usar pgvector#

Próximos passos#

Resumo prático#

Continue lendo

Chunking: as estratégias que funcionam

RAG na prática: do zero ao primeiro retriever

RAG, fine-tuning ou prompt? Guia de decisão

O problema real

Por que pgvector cobre a maioria dos casos?

Mão na massa

Setup: instalar pgvector

Criar tabela com coluna vetorial

Inserir embeddings

Busca por similaridade (cosseno)

Busca híbrida (vetorial + full-text)

Filtros de metadados durante busca

Armadilhas comuns

Armadilha 1: Índice HNSW vs IVFFlat

Armadilha 2: Não normalizar embeddings

Armadilha 3: Forget to vacuum

Quando não usar pgvector

Próximos passos

Resumo prático