Redis®*: Python範例
本文件是向量搜尋與RAG指南的一部分。請在此處查看完整指南：使用Redis進行語意搜尋、推薦系統與檢索增強生成（RAG）。
👋 歡迎來到 Stackhero 文件中心！
Stackhero 提供即時可用的 Redis cloud 解決方案，帶來多項優勢，包括：
- 內建
Redis Commander網頁管理介面。- 無限制的訊息大小與傳輸量。
- 一鍵輕鬆完成更新。
- 透過專屬私有基礎架構，實現最佳效能與強大安全性。
節省時間、簡化您的工作流程：只需 5 分鐘即可體驗 Stackhero 的 Redis cloud hosting 解決方案！
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_REDIS_URL_TLS'])
# 啟動時建立索引（僅需一次）
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 AS score]')
.sort_by('score')
.return_fields('content', 'score')
.dialect(2)
)
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]