Valkey: La même chose en Python
Cette documentation fait partie du guide Recherche vectorielle et RAG. Consultez le guide complet ici : Recherche sémantique, recommandations et retrieval-augmented generation avec Valkey.
👋 Bienvenue dans la documentation Stackhero !
Stackhero propose une solution Valkey cloud clé en main qui offre de nombreux avantages, dont :
- Interface web Valkey Admin incluse.
- Taille et transferts de messages illimités.
- Mises à jour faciles en un seul clic.
- Performance optimale et sécurité avancée grâce à une infrastructure privée et dédiée.
Gagnez du temps et simplifiez-vous la vie : il ne vous faut que 5 minutes pour essayer l’hébergement Valkey cloud de Stackhero !
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'])
# Créez l'index une seule fois, au démarrage
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]