Valkey: ベクトル検索とRAG

Valkeyによるセマンティック検索、レコメンデーション、リトリーバル拡張生成（RAG）

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

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

  • Valkey Admin Web UI を標準搭載。
  • メッセージサイズ・転送量が無制限。
  • ワンクリックで簡単にアップデート可能。
  • プライベート専用インフラによる最適なパフォーマンスと強固なセキュリティ。

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

セマンティック検索、レコメンデーションエンジン、またはリトリーバル拡張生成（RAG）パイプラインを構築する場合、埋め込み（embedding）を保存し、クエリに対して最も近いマッチを高速に検索する必要があります。Stackhero for Valkeyはこれをネイティブにサポートしています。検索モジュールがベクトルをインデックスし、類似度検索にミリ秒単位で応答します。既存のValkeyサービスの横に別途ベクトルデータベースを用意する必要はありません。

このアプローチを実用的かつ効率的にしているのは、次の2つの特徴です：

  • ハイブリッドクエリ。 ベクトル類似度検索と通常のフィルタ（数値範囲、タグ、テキストなど）を1つのリクエストで組み合わせて実行できます。たとえば、「この質問に最も近い5つのチャンクを取得するが、このユーザーが閲覧可能なドキュメントのみ対象とする」といった条件も1回のクエリで実現できます。
  • 構成要素の削減。 埋め込みデータはキャッシュやセッションデータと同じ場所に保存され、同じ認証情報、バックアップ、監視ツールを利用できます。これによりスタックがシンプルになり、運用管理が容易になります。

はじめに

Stackheroダッシュボードのサービス設定画面「Modules」セクションで、Searchモジュール（ドキュメントをJSON形式で保存したい場合はJSONも）を有効にしてください。詳細はsearchとJSONガイドをご参照ください。

ベクトルインデックスの作成

ベクトルフィールドを宣言する際は、その次元数（dimension）と距離メトリック（distance metric）を指定します。次元数はご利用のモデルに合わせてください。例：OpenAIのtext-embedding-3-smallなら1536、多くのオープンソースモデルなら768など。

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はグラフインデックスを作成し、コレクションが大規模になっても高速な近似検索が可能です。数千ベクトル以上のコレクションにはこれを推奨します。小規模なコレクションの場合は、FLATによる厳密な全探索（brute-force）検索も利用できます。

テキスト埋め込みには通常COSINE距離が最適です。L2やIPメトリックもサポートされています。

クエリの実行

K近傍（KNN）クエリは、ベクトルをリトルエンディアンのfloat32バイト列としてパラメータで渡します：

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

ベクトル検索とフィルタを組み合わせる場合は、*をフィルタ式に置き換えるだけです：

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

Node.jsによるRAGパイプライン

ValkeyはRedisと同じプロトコルを使用しているため、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. インデックスは起動時に一度だけ作成します
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. チャンクをインデックス化（embeddingはモデルから取得）
async function indexChunk({ id, documentId, content, embedding }) {
  await client.hSet(`chunk:${id}`, {
    content,
    documentId,
    embedding: toBytes(embedding)
  });
}

// 3. 質問に最も近いチャンクを、特定ドキュメントに限定して取得
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. モデルへのコンテキストとして渡す
const context = await retrieve({ questionEmbedding, documentId: 'doc42' });
const prompt = `Answer using only this context:\n\n${context.join('\n\n')}\n\nQuestion: ${question}`;

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

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

サービスのサイズ設計

ベクトルはメモリ上に保存されるため、そのメモリフットプリントを事前に見積もることが重要です。float32ベクトルの場合、おおよその計算式は：

numberOfVectors x dimensions x 4 bytes に加え、HNSWグラフ用に約30～50%の追加メモリが必要です。

例えば、1536次元のベクトルが100万件ある場合、生データだけで約6GB必要です。20GBプランであれば余裕を持って運用できます。次元数の少ないモデルを選択すれば、メモリ使用量を削減できます（768次元モデルなら1536次元の半分のメモリで済みます）。

実際のメモリ使用量は FT.INFO chunksIndex や、Prometheus監視の used_memory メトリクスで確認できます。必要に応じてダッシュボードからいつでもプランをアップグレード可能です。

補足情報

  • 既存キーのインデックス化はバックグラウンドで実行されます。 進捗は FT.INFO で確認できます。バックフィルが完了するまでは検索結果が部分的になる場合があります。
  • 検索モジュールは厳選されたFT.*コマンドセットを実装しています。 ベクトル検索に必要な機能（HNSW・厳密なKNN・ハイブリッドフィルタ・FT.AGGREGATE）はすべてカバーしています。

詳細はValkey searchドキュメントをご参照ください。

Valkey に関するその他の記事

  1. はじめに
    Valkeyのクイックスタートガイド
  2. プロメテウス
    プロメテウスのメトリクス
  3. Pythonでの使用
    ValkeyをPythonで接続する方法
  4. Rubyでの利用方法
    RubyでValkeyに接続する方法
  5. 検索とJSON
    Valkeyの検索モジュールでJSONドキュメントを保存・検索する
  6. ベクトル検索とRAG
    Valkeyによるセマンティック検索、レコメンデーション、リトリーバル拡張生成（RAG）
  7. CLIによる自動化
    Stackhero CLIを使ってValkeyを起動し、認証情報を取得し、構成をプログラムで変更する方法
  8. GitHub Actions & GitLab CI
    GitHub Actions または GitLab CI パイプラインから実際の Valkey サービスを起動し、テストを実行して、自動的に削除する方法