Valkey: Pesquisa vetorial e RAG

Pesquisa semântica, recomendações e retrieval-augmented generation com Valkey

👋 Bem-vindo à documentação da Stackhero!

A Stackhero disponibiliza uma solução Valkey cloud pronta a usar, que oferece várias vantagens, incluindo:

• Interface web Valkey Admin incluída.
• Tamanho e transferências de mensagens ilimitados.
• Atualizações fáceis com apenas um clique.
• Performance otimizada e segurança reforçada graças a uma infraestrutura privada e dedicada.

Poupe tempo e simplifique a sua vida: bastam 5 minutos para experimentar a solução de Valkey cloud hosting da Stackhero!

Se está a desenvolver pesquisa semântica, um motor de recomendações ou um pipeline de retrieval-augmented generation (RAG), precisa de armazenar embeddings e encontrar rapidamente as correspondências mais próximas para uma consulta. O Stackhero para Valkey trata disto de forma nativa: o módulo de pesquisa indexa vetores e responde a consultas de similaridade em milissegundos. Não é necessário configurar uma base de dados vetorial separada além do serviço Valkey que já utiliza.

Duas funcionalidades tornam esta abordagem prática e eficiente:

• Consultas híbridas. Pode executar pesquisas de similaridade vetorial e aplicar filtros tradicionais (como intervalos numéricos, tags ou texto) num único pedido. Por exemplo, pode obter "os 5 segmentos mais próximos desta pergunta, mas apenas de documentos que este utilizador tem permissão para ler" com uma só consulta, sem necessidade de duas.
• Menos componentes para gerir. Os seus embeddings são armazenados juntamente com os dados de cache e de sessão, utilizando as mesmas credenciais, backups e ferramentas de monitorização. Isto simplifica a sua stack e facilita a gestão.

Antes de começar

Ative o módulo Search (e JSON se quiser armazenar documentos em formato JSON) na secção Modules da configuração do seu serviço no dashboard da Stackhero. Para mais detalhes, consulte o guia de search e JSON.

Criar um índice vetorial

Declare um campo vetorial especificando a sua dimensão e a métrica de distância. A dimensão deve corresponder ao seu modelo: por exemplo, 1536 para o OpenAI text-embedding-3-small, ou 768 para muitos modelos open source.

FT.CREATE chunksIndex
  ON HASH
  PREFIX 1 chunk:
  SCHEMA
    content TEXT
    documentId TAG
    embedding VECTOR HNSW 6
      TYPE FLOAT32
      DIM 1536
      DISTANCE_METRIC COSINE

HNSW cria um índice em grafo, permitindo pesquisas aproximadas rápidas mesmo quando a coleção cresce. É recomendado para coleções superiores a alguns milhares de vetores. Para coleções pequenas, pode usar FLAT para pesquisa exata e exaustiva.

COSINE é normalmente a melhor métrica de distância para embeddings de texto. As métricas L2 e IP também são suportadas.

Consultas

Uma consulta K-nearest neighbors (KNN) passa o vetor como parâmetro, usando bytes float32 little-endian brutos:

FT.SEARCH chunksIndex "*=>[KNN 5 @embedding $queryVector]"
  PARAMS 2 queryVector "<rawBytes>"

Para combinar pesquisa vetorial com filtros, basta substituir o * pela sua expressão de filtro:

FT.SEARCH chunksIndex "@documentId:{doc42}=>[KNN 5 @embedding $queryVector]"
  PARAMS 2 queryVector "<rawBytes>"

Um pipeline RAG em Node.js

O Valkey utiliza o mesmo protocolo que o Redis, por isso pode usar qualquer cliente compatível com Redis.

import { createClient, SCHEMA_FIELD_TYPE, SCHEMA_VECTOR_FIELD_ALGORITHM } from 'redis';

const client = createClient({ url: process.env.STACKHERO_VALKEY_URL_TLS });
await client.connect();

const DIMENSIONS = 1536;
const toBytes = embedding => Buffer.from(new Float32Array(embedding).buffer);

// 1. Criar o índice uma vez, no arranque
try {
  await client.ft.create(
    'chunksIndex',
    {
      content: SCHEMA_FIELD_TYPE.TEXT,
      documentId: SCHEMA_FIELD_TYPE.TAG,
      embedding: {
        type: SCHEMA_FIELD_TYPE.VECTOR,
        ALGORITHM: SCHEMA_VECTOR_FIELD_ALGORITHM.HNSW,
        TYPE: 'FLOAT32',
        DIM: DIMENSIONS,
        DISTANCE_METRIC: 'COSINE'
      }
    },
    { ON: 'HASH', PREFIX: 'chunk:' }
  );
}
catch (error) {
  if (!error.message.includes('Index already exists')) {
    throw error;
  }
}

// 2. Indexar os seus segmentos (embedding fornecido pelo seu modelo)
async function indexChunk({ id, documentId, content, embedding }) {
  await client.hSet(`chunk:${id}`, {
    content,
    documentId,
    embedding: toBytes(embedding)
  });
}

// 3. Recuperar os segmentos mais próximos de uma pergunta, restritos a um documento
async function retrieve({ questionEmbedding, documentId, count = 5 }) {
  const results = await client.ft.search(
    'chunksIndex',
    `@documentId:{${documentId}}=>[KNN ${count} @embedding $queryVector]`,
    { PARAMS: { queryVector: toBytes(questionEmbedding) } }
  );

  return results.documents.map(({ value }) => value.content);
}

// 4. Forneça-os ao seu modelo como contexto
const context = await retrieve({ questionEmbedding, documentId: 'doc42' });
const prompt = `Responda apenas usando este contexto:\n\n${context.join('\n\n')}\n\nPergunta: ${question}`;

O mesmo em Python

import os
import numpy as np
import redis
from redis.commands.search.field import TextField, TagField, VectorField
from redis.commands.search.index_definition import IndexDefinition, IndexType
from redis.commands.search.query import Query

DIMENSIONS = 1536

r = redis.from_url(os.environ['STACKHERO_VALKEY_URL_TLS'])

# Criar o índice uma vez, no arranque
try:
    r.ft('chunksIndex').create_index(
        (
            TextField('content'),
            TagField('documentId'),
            VectorField(
                'embedding',
                'HNSW',
                {'TYPE': 'FLOAT32', 'DIM': DIMENSIONS, 'DISTANCE_METRIC': 'COSINE'},
            ),
        ),
        definition=IndexDefinition(prefix=['chunk:'], index_type=IndexType.HASH),
    )
except redis.ResponseError as error:
    if 'Index already exists' not in str(error):
        raise


def index_chunk(chunk_id, document_id, content, embedding):
    r.hset(
        f'chunk:{chunk_id}',
        mapping={
            'content': content,
            'documentId': document_id,
            'embedding': np.array(embedding, dtype=np.float32).tobytes(),
        },
    )


def retrieve(question_embedding, document_id, count=5):
    query = Query(f'@documentId:{{{document_id}}}=>[KNN {count} @embedding $queryVector]')
    results = r.ft('chunksIndex').search(
        query,
        query_params={'queryVector': np.array(question_embedding, dtype=np.float32).tobytes()},
    )
    return [document.content for document in results.docs]

Dimensionar o seu serviço

Os vetores são armazenados em memória, por isso é importante planear o seu impacto. A estimativa aproximada para vetores float32 é:

numberOfVectors x dimensions x 4 bytes, mais cerca de 30 a 50 por cento extra para o grafo HNSW.

Por exemplo, um milhão de vetores com 1536 dimensões requerem cerca de 6 GB para os vetores brutos. Um plano de 20 GB é um ponto de partida confortável. Pode reduzir o uso de memória escolhendo um modelo com menos dimensões: um modelo de 768 dimensões utiliza apenas metade da memória de um de 1536 dimensões.

Pode verificar o consumo real de memória com FT.INFO chunksIndex e com a métrica used_memory na sua monitorização Prometheus. Se precisar de mais memória, pode atualizar o seu plano a qualquer momento a partir do dashboard.

Informação útil

• A indexação de chaves existentes é feita em background. Pode acompanhar o progresso com FT.INFO. As pesquisas devolvem resultados parciais até que o preenchimento esteja concluído.
• O módulo de pesquisa implementa um conjunto focado de comandos FT.*. Para pesquisa vetorial, cobre o essencial: HNSW e KNN exato, filtragem híbrida e FT.AGGREGATE.

Para referência completa, consulte a documentação Valkey search.