Valkey: Búsqueda vectorial y RAG

Búsqueda semántica, recomendaciones y retrieval-augmented generation con Valkey

👋 ¡Bienvenido a la documentación de Stackhero!

Stackhero ofrece una solución Valkey cloud lista para usar que proporciona numerosas ventajas, entre ellas:

• Interfaz web Valkey Admin incluida.
• Tamaño y transferencia de mensajes ilimitados.
• Actualizaciones sencillas con solo un clic.
• Rendimiento óptimo y seguridad reforzada gracias a una infraestructura privada y dedicada.

Ahorre tiempo y simplifique su trabajo: solo necesita 5 minutos para probar la solución de Valkey cloud hosting de Stackhero.

Si está desarrollando una búsqueda semántica, un motor de recomendaciones o un pipeline de retrieval-augmented generation (RAG), necesita almacenar embeddings y encontrar rápidamente las coincidencias más cercanas para una consulta. Stackhero para Valkey gestiona esto de forma nativa: el módulo de búsqueda indexa los vectores y responde a consultas de similitud en milisegundos. No es necesario configurar una base de datos vectorial independiente junto al servicio Valkey que ya utiliza.

Dos características hacen que este enfoque sea práctico y eficiente:

• Consultas híbridas. Puede realizar búsquedas de similitud vectorial y aplicar filtros convencionales (como rangos numéricos, etiquetas o texto) en una sola petición. Por ejemplo, puede recuperar "los 5 fragmentos más cercanos a esta pregunta, pero solo de documentos que este usuario tiene permiso para leer" con una sola consulta, no dos.
• Un componente menos en su arquitectura. Sus embeddings se almacenan junto con los datos de caché y sesión, utilizando las mismas credenciales, copias de seguridad y herramientas de monitorización. Esto mantiene su stack más simple y fácil de gestionar.

Antes de empezar

Active el módulo Search (y JSON si desea almacenar documentos en formato JSON) en la sección Modules de la configuración de su servicio en el panel de Stackhero. Para más detalles, consulte la guía de search y JSON.

Creación de un índice vectorial

Declare un campo vectorial especificando su dimensión y la métrica de distancia. La dimensión debe coincidir con la de su modelo: por ejemplo, 1536 para OpenAI text-embedding-3-small, o 768 para muchos 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 crea un índice basado en grafos, permitiendo búsquedas aproximadas rápidas incluso cuando la colección crece mucho. Se recomienda para colecciones de más de unos pocos miles de vectores. Para colecciones pequeñas, puede usar FLAT para una búsqueda exacta y exhaustiva.

COSINE suele ser la mejor métrica de distancia para embeddings de texto. También se admiten las métricas L2 e IP.

Consultas

Una consulta K-nearest neighbors (KNN) pasa el vector como parámetro, usando bytes float32 en little-endian sin procesar:

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

Para combinar búsqueda vectorial con filtros, simplemente reemplace el * por su expresión de filtro:

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

Un pipeline RAG en Node.js

Valkey utiliza el mismo protocolo que Redis, por lo que puede usar cualquier cliente compatible con 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. Cree el índice una sola vez, al iniciar
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. Indexe sus fragmentos (el embedding lo proporciona su modelo)
async function indexChunk({ id, documentId, content, embedding }) {
  await client.hSet(`chunk:${id}`, {
    content,
    documentId,
    embedding: toBytes(embedding)
  });
}

// 3. Recupere los fragmentos más cercanos a una pregunta, restringidos a un 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. Úselos como contexto para su modelo
const context = await retrieve({ questionEmbedding, documentId: 'doc42' });
const prompt = `Responde utilizando solo este contexto:\n\n${context.join('\n\n')}\n\nPregunta: ${question}`;

Lo mismo en 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'])

# Cree el índice una sola vez, al iniciar
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 su servicio

Los vectores se almacenan en memoria, por lo que es importante planificar su huella. La estimación aproximada para vectores float32 es:

numberOfVectors x dimensions x 4 bytes, más aproximadamente entre un 30 y un 50 por ciento adicional para el grafo HNSW.

Por ejemplo, un millón de vectores de 1536 dimensiones requieren unos 6 GB para los vectores en bruto. Un plan de 20 GB es un punto de partida cómodo. Puede reducir el uso de memoria eligiendo un modelo con menos dimensiones: un modelo de 768 dimensiones utiliza solo la mitad de memoria que uno de 1536 dimensiones.

Puede comprobar el uso real de memoria con FT.INFO chunksIndex y con la métrica used_memory en su monitorización Prometheus. Si necesita más memoria, puede ampliar su plan en cualquier momento desde su panel de control.

Información útil

• La indexación de claves existentes se realiza en segundo plano. Puede seguir el progreso con FT.INFO. Las búsquedas devolverán resultados parciales hasta que finalice el proceso de backfilling.
• El módulo de búsqueda implementa un conjunto específico de comandos FT.*. Para la búsqueda vectorial, cubre lo esencial: HNSW y KNN exacto, filtrado híbrido y FT.AGGREGATE.

Para una referencia completa, consulte la documentación de búsqueda de Valkey.