Redis®*: 向量搜索与RAG

使用Redis实现语义搜索、推荐系统和检索增强生成（RAG）

👋 欢迎查阅 Stackhero 文档！

Stackhero 提供即开即用的 Redis cloud 解决方案，带来多项优势，包括：

集成 Redis Commander Web 管理界面。

无限制的消息大小与传输。

一键完成升级，操作便捷。

基于专属私有基础设施，实现卓越性能与强大安全性。

节省时间，简化运维：仅需 5 分钟即可体验 Stackhero 的 Redis cloud 托管方案！

如果您正在构建语义搜索、推荐引擎或检索增强生成（RAG）流程，您需要一种可靠的方式来存储embedding，并能够快速找到与查询最接近的匹配项。您在Stackhero上的Redis实例原生支持这一功能：查询引擎会对向量进行索引，并能在毫秒级别响应相似度搜索。无需为Redis单独添加向量数据库：所有功能都集中在同一处运行。

这种方案之所以高效且易用，得益于两个关键特性：

混合查询。 您可以在一次请求中，将向量相似度搜索与常规过滤（如数值区间、标签或全文检索）结合。例如：“找出与该问题最接近的5个片段，但仅限于该用户有权限阅读、且发布时间在一月之后的文档”，这一需求可以通过单次查询完成，无需多次调用。
简化架构。 embedding、缓存和会话全部集中管理，共享凭据、备份和监控。这大大降低了系统复杂度和运维成本。

开始之前

请在Stackhero 控制台的服务配置 Modules 部分启用 Search 模块（如需以JSON格式存储文档，还需启用 JSON 模块）。详细步骤请参阅Search与JSON指南。

创建向量索引

声明向量字段时，需要指定其维度和距离度量方式。维度应与您的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. 将检索结果作为上下文传递给您的模型
const context = await retrieve({ questionEmbedding, documentId: 'doc42' });
const prompt = `请仅使用以下上下文回答：\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，再加上HNSW图结构大约30–50%的额外空间。

例如，一百万条1536维的向量，原始向量大约占用6GB内存。对于这一规模，建议选择20GB的套餐作为起步。您可以通过以下方式进一步降低内存占用：

使用更小的模型。 例如，768维模型的内存消耗仅为1536维模型的一半。
如模型支持，可将存储类型由 FLOAT32 改为 FLOAT16，再次将空间减半。

您可以通过 FT.INFO chunksIndex 和 Prometheus监控中的 used_memory 指标实时监控实际使用情况。如需扩容，可直接在控制台升级套餐，操作便捷。

温馨提示

向量索引仅支持数据库0，与Redis所有搜索索引一致。
向量集合（Vector sets）提供了无需配置的简化方案。 Redis内置了 VADD 和 VSIM 命令，无需启用任何模块即可立即进行相似度搜索。适用于单集合的简单推荐场景。如需过滤、分页或多字段查询，建议使用查询引擎。
对已有键构建索引会在后台异步进行。 可通过 FT.INFO 查看进度。在索引完成前，搜索会返回部分结果。

如需了解更多关于Redis向量搜索的信息，请参阅官方Redis向量搜索文档。