Valkey: 向量搜尋與 RAG

使用 Valkey 進行語意搜尋、推薦系統及檢索增強生成（Retrieval-Augmented Generation）

👋 歡迎瀏覽 Stackhero 文件！

Stackhero 提供即開即用的 Valkey cloud 方案，帶來多項優勢，包括：

已包含 Valkey Admin 網頁管理介面。

無限訊息大小及傳輸量。

一鍵輕鬆完成更新。

以專屬私有基礎設施提供最佳效能及強大安全性。

節省時間，簡化您的工作流程：只需 5 分鐘即可體驗 Stackhero 的 Valkey cloud hosting 方案！

如果您正在開發語意搜尋、推薦引擎或 Retrieval-Augmented Generation（RAG）流程，您需要儲存 embeddings，並能夠快速找出與查詢最接近的匹配。Stackhero for Valkey 原生支援這些需求：搜尋模組會為向量建立索引，並能在毫秒內回應相似度查詢。您無需額外部署獨立的向量資料庫，只需使用現有的 Valkey 服務即可。

這種做法之所以實用且高效，主要有兩個原因：

混合查詢（Hybrid queries）。 您可以在單一請求中同時進行向量相似度搜尋，並套用一般篩選條件（如數值範圍、標籤或文字）。例如，您可以用一個查詢就取得「最接近這條問題的 5 個片段，但僅限於這位用戶有權閱讀的文件」，無需分兩次查詢。
減少系統組件。 您的 embeddings 會與快取和 session 資料一併儲存，使用相同的認證、備份及監控工具。這樣能讓您的技術堆疊更簡單，管理起來也更容易。

開始之前

請在 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 進行精確、暴力搜尋。

COSINE 通常是文本 embedding 最佳的距離度量方式。系統同時支援 L2 及 IP。

查詢

K-nearest neighbors（KNN）查詢會以參數方式傳遞向量，格式為原始 little-endian float32 bytes：

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 相容的 client。

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. 將片段（chunk）建立索引（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. 將這些片段作為 context 傳給您的模型
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 graph 約 30% 至 50% 的額外空間。

例如，一百萬個 1536 維的向量，僅原始向量就需約 6GB。建議選擇 20GB 的方案作為起步。若選用維度較少的模型（如 768 維），記憶體用量僅為 1536 維模型的一半。

您可透過 FT.INFO chunksIndex 及 Prometheus 監控的 used_memory 指標查看實際記憶體用量。如需更多記憶體，隨時可於控制台升級方案。

實用資訊

現有 key 的索引會於背景執行。 您可用 FT.INFO 追蹤進度。在回填完成前，搜尋結果可能僅為部分資料。
搜尋模組僅實作精選的 FT.* 指令。 向量搜尋涵蓋所有重點功能：HNSW 及精確 KNN、混合篩選及 FT.AGGREGATE。

完整說明請參閱 Valkey search 官方文件。