Redis®*: 向量搜尋與RAG

使用Redis進行語意搜尋、推薦系統與檢索增強生成（RAG）

👋 歡迎來到 Stackhero 文件中心！

Stackhero 提供即時可用的 Redis cloud 解決方案，帶來多項優勢，包括：

內建 Redis Commander 網頁管理介面。

無限制的訊息大小與傳輸量。

一鍵輕鬆完成更新。

透過專屬私有基礎架構，實現最佳效能與強大安全性。

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

如果您正在建構語意搜尋、推薦引擎或檢索增強生成（RAG）流程，您需要一個可靠的方式來儲存embeddings，並能快速找到與查詢最接近的匹配。您在Stackhero上的Redis實例原生支援這項功能：查詢引擎會索引向量，並在毫秒內回應相似度搜尋。您不需要額外部署獨立的向量資料庫，所有功能都集中在同一個地方。

這種做法之所以高效且實用，主要歸功於兩個關鍵特性：

混合查詢（Hybrid queries）。 您可以將向量相似度搜尋與標準篩選條件（如數值範圍、標籤或全文檢索）結合在同一個請求中。例如：「找出最接近這個問題的5個片段，但僅限於這位使用者有權閱讀、且發佈時間在一月之後的文件」——這樣的需求可以用一個查詢完成，無需多次請求。
簡化架構。 Embeddings、快取與Session都集中管理，共用認證、備份與監控，降低了系統複雜度與維運負擔。

開始之前

請在Stackhero 控制台的服務設定「Modules」區段啟用 Search 模組（若您打算以JSON格式儲存文件，也請啟用 JSON 模組）。詳細步驟請參考Search與JSON教學。

建立向量索引

要宣告一個向量欄位，請指定其維度（dimension）與距離度量（distance metric）。維度需與您的embedding模型相符：例如，OpenAI text-embedding-3-small為1536，許多開源模型則為768，依此類推。

FT.CREATE chunksIndex
  ON HASH
  PREFIX 1 chunk:
  SCHEMA
    content TEXT
    documentId TAG
    publishedAt NUMERIC
    embedding VECTOR HNSW 6
      TYPE FLOAT32
      DIM 1536
      DISTANCE_METRIC COSINE

HNSW 會建立一個圖形索引，讓查詢速度隨著資料量成長仍能維持高效，適合超過數千筆向量的情境。若您的資料集較小且重視精確結果，可以選擇 FLAT 進行暴力搜尋。

COSINE 是大多數文本embedding模型推薦的距離度量。若您的應用需求不同，也可選擇 L2 或 IP。

查詢

K近鄰（KNN）查詢範例如下。向量需以原始little-endian float32位元組格式傳入：

FT.SEARCH chunksIndex "*=>[KNN 5 @embedding $queryVector AS score]"
  PARAMS 2 queryVector "<rawBytes>"
  SORTBY score
  RETURN 2 content score
  DIALECT 2

若要結合相似度搜尋與篩選條件，只需將 * 替換為篩選表達式：

FT.SEARCH chunksIndex "(@documentId:{doc42} @publishedAt:[1735689600 +inf])=>[KNN 5 @embedding $queryVector AS score]"
  PARAMS 2 queryVector "<rawBytes>"
  SORTBY score
  DIALECT 2

向量查詢必須指定 DIALECT 2。若省略此參數，查詢會採用舊語法並回傳錯誤。

Node.js中的RAG流程範例

import { createClient, SCHEMA_FIELD_TYPE, SCHEMA_VECTOR_FIELD_ALGORITHM } from 'redis';

const client = createClient({ url: process.env.STACKHERO_REDIS_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 AS score]`,
    {
      PARAMS: { queryVector: toBytes(questionEmbedding) },
      SORTBY: 'score',
      RETURN: [ 'content', 'score' ],
      DIALECT: 2
    }
  );

  return results.documents.map(({ value }) => value.content);
}

// 4. 將查詢結果作為context傳給您的模型
const context = await retrieve({ questionEmbedding, documentId: 'doc42' });
const prompt = `請僅根據以下context回答：\n\n${context.join('\n\n')}\n\n問題：${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_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]

服務規模評估

向量資料會儲存在記憶體中。您可以用以下公式預估所需記憶體：

numberOfVectors x dimensions x 4 bytes，再加上約30–50%作為HNSW圖形結構的額外空間。

例如，一百萬筆1536維的向量，僅原始向量就需約6GB記憶體。20GB的方案對這個規模來說是個不錯的起點。您可以透過以下方式降低記憶體用量：

使用較小的模型。 例如，768維的模型比1536維的模型節省一半記憶體。
若模型支援，將FLOAT32改為FLOAT16，可再減半空間需求。

您可以透過 FT.INFO chunksIndex 及 Prometheus監控中的 used_memory 指標監控實際用量。若用量增加，您可直接在控制台升級方案，流程簡單。

實用須知

向量索引僅支援資料庫0，這與Redis所有搜尋索引一致。
向量集合（Vector sets）提供更簡單的替代方案，無需額外設定。 Redis內建 VADD 與 VSIM 指令，讓您可立即進行相似度搜尋，無需啟用模組。這對於單一集合的簡單推薦場景非常方便。若您需要篩選、分頁或多欄位查詢，建議使用查詢引擎。
針對現有鍵值建立索引會在背景執行。 您可用 FT.INFO 查詢進度。索引尚未完成時，搜尋會回傳部分結果。

如需進一步瞭解Redis向量搜尋，請參考官方Redis向量搜尋文件。