Valkey: Pythonでの実装例

このドキュメントはベクトル検索とRAGガイドの一部です。完全なガイドはこちらからご覧いただけます：Valkeyによるセマンティック検索、レコメンデーション、リトリーバル拡張生成（RAG）。

👋 Stackhero ドキュメントへようこそ！

Stackhero では、すぐに利用可能な Valkey クラウドソリューションを提供しています。主な特長は以下の通りです：

Valkey Admin Web UI を標準搭載。

メッセージサイズ・転送量が無制限。

ワンクリックで簡単にアップデート可能。

プライベート専用インフラによる最適なパフォーマンスと強固なセキュリティ。

時間を節約し、運用をシンプルに：Stackhero の Valkey クラウドホスティングソリューションは、わずか5分でお試しいただけます！

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'])

# インデックスは起動時に一度だけ作成します
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]

Valkey: Pythonでの実装例

👋 Stackhero ドキュメントへようこそ！

このガイドの続きをお読みください

Valkey に関するその他の記事