Valkey: Ricerca vettoriale e RAG

Ricerca semantica, raccomandazioni e retrieval-augmented generation con Valkey

👋 Benvenuto nella documentazione di Stackhero!

Stackhero offre una soluzione Valkey cloud pronta all'uso che garantisce numerosi vantaggi, tra cui:

• Interfaccia web Valkey Admin inclusa.
• Dimensione e trasferimento dei messaggi illimitati.
• Aggiornamenti semplici con un solo clic.
• Prestazioni ottimali e sicurezza avanzata grazie a un'infrastruttura privata e dedicata.

Risparmia tempo e semplificati la vita: bastano 5 minuti per provare la soluzione Valkey cloud hosting di Stackhero!

Se state sviluppando una ricerca semantica, un motore di raccomandazione o una pipeline di retrieval-augmented generation (RAG), è necessario memorizzare gli embeddings e trovare rapidamente le corrispondenze più vicine per una query. Stackhero per Valkey gestisce questa esigenza in modo nativo: il modulo di ricerca indicizza i vettori e risponde alle query di similarità in pochi millisecondi. Non è necessario configurare un database vettoriale separato oltre al servizio Valkey che già utilizzate.

Due funzionalità rendono questo approccio pratico ed efficiente:

• Query ibride. È possibile eseguire ricerche di similarità vettoriale e applicare filtri tradizionali (come intervalli numerici, tag o testo) in un'unica richiesta. Ad esempio, potete recuperare "i 5 segmenti più vicini a questa domanda, ma solo dai documenti che questo utente è autorizzato a leggere" con una sola query, non due.
• Un componente in meno da gestire. Gli embeddings vengono memorizzati insieme ai dati di cache e sessione, utilizzando le stesse credenziali, backup e strumenti di monitoraggio. Questo mantiene lo stack più semplice e facile da gestire.

Prima di iniziare

Abilitate il modulo Search (e JSON se desiderate memorizzare i documenti in formato JSON) nella sezione Modules della configurazione del vostro servizio sulla dashboard Stackhero. Per maggiori dettagli, consultate la guida search e JSON.

Creazione di un indice vettoriale

Dichiarate un campo vettoriale specificando la dimensione e la metrica di distanza. La dimensione deve corrispondere al vostro modello: ad esempio, 1536 per OpenAI text-embedding-3-small, oppure 768 per molti modelli 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 indice a grafo, offrendo ricerche approssimative rapide anche quando la collezione cresce molto. È consigliato per collezioni superiori a qualche migliaio di vettori. Per collezioni piccole, potete usare FLAT per una ricerca esatta e brute-force.

COSINE è solitamente la metrica di distanza migliore per gli embeddings di testo. Sono supportate anche le metriche L2 e IP.

Query

Una query K-nearest neighbors (KNN) passa il vettore come parametro, utilizzando byte float32 little-endian grezzi:

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

Per combinare la ricerca vettoriale con i filtri, basta sostituire * con la vostra espressione di filtro:

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

Una pipeline RAG in Node.js

Valkey utilizza lo stesso protocollo di Redis, quindi potete usare qualsiasi client compatibile 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. Create l'indice una sola volta, all'avvio
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. Indicizzate i vostri segmenti (embedding fornito dal vostro modello)
async function indexChunk({ id, documentId, content, embedding }) {
  await client.hSet(`chunk:${id}`, {
    content,
    documentId,
    embedding: toBytes(embedding)
  });
}

// 3. Recuperate i segmenti più vicini a una domanda, limitati 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. Fornite questi segmenti al vostro modello come contesto
const context = await retrieve({ questionEmbedding, documentId: 'doc42' });
const prompt = `Rispondi utilizzando solo questo contesto:\n\n${context.join('\n\n')}\n\nDomanda: ${question}`;

Lo stesso in 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'])

# Create l'indice una sola volta, all'avvio
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]

Dimensionamento del servizio

I vettori vengono memorizzati in memoria, quindi è importante pianificare il loro impatto. Una stima approssimativa per vettori float32 è:

numberOfVectors x dimensions x 4 bytes, più circa il 30-50% in più per il grafo HNSW.

Ad esempio, un milione di vettori da 1536 dimensioni richiedono circa 6 GB per i vettori grezzi. Un piano da 20 GB rappresenta un punto di partenza confortevole. Potete ridurre l'utilizzo di memoria scegliendo un modello con meno dimensioni: un modello da 768 dimensioni utilizza solo la metà della memoria rispetto a uno da 1536.

Potete verificare l'utilizzo reale della memoria con FT.INFO chunksIndex e con la metrica used_memory nel vostro monitoraggio Prometheus. Se avete bisogno di più memoria, potete aggiornare il vostro piano in qualsiasi momento dalla dashboard.

Da sapere

• L'indicizzazione delle chiavi esistenti avviene in background. Potete monitorare l'avanzamento con FT.INFO. Le ricerche restituiranno risultati parziali fino al completamento del backfilling.
• Il modulo di ricerca implementa un set mirato di comandi FT.*. Per la ricerca vettoriale, copre tutte le funzionalità essenziali: HNSW e KNN esatto, filtraggio ibrido e FT.AGGREGATE.

Per la documentazione completa, consultate la documentazione Valkey search.