Valkey: 向量搜索与RAG

使用Valkey进行语义搜索、推荐和检索增强生成（RAG）

👋 欢迎来到 Stackhero 文档！

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

内置 Valkey Admin Web 管理界面。

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

只需一键即可轻松完成更新。

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

节省时间，简化您的工作流程：只需 5 分钟即可体验 Stackhero 的 Valkey cloud hosting 解决方案！

如果您正在构建语义搜索、推荐引擎或检索增强生成（RAG）流程，您需要存储embedding，并能够快速找到与查询最接近的匹配项。Stackhero for Valkey原生支持这一需求：Search模块会对向量进行索引，并能在毫秒级别响应相似度查询。您无需为已在使用的Valkey服务额外部署独立的向量数据库。

这种方案之所以高效实用，主要得益于以下两大特性：

混合查询。 您可以在一次请求中同时进行向量相似度搜索，并应用常规过滤条件（如数值范围、标签或文本）。例如，您可以通过单次查询检索“与该问题最接近的5个片段，但仅限于该用户有权限阅读的文档”，无需分两步操作。
减少系统复杂度。 您的embedding与缓存和会话数据一同存储，使用相同的凭证、备份和监控工具。这让您的技术栈更简洁，管理也更轻松。

开始之前

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

创建向量索引

声明向量字段时需指定其维度和距离度量方式。维度需与您的模型一致：例如，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近邻（KNN）查询通过参数传递向量，格式为原始little-endian 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维向量的原始数据约需6GB内存。建议选择20GB套餐作为起步。若选用维度更低的模型（如768维），内存占用仅为1536维模型的一半。

您可以通过FT.INFO chunksIndex命令和Prometheus监控中的used_memory指标查看实际内存使用情况。如需扩容，可随时在控制台升级套餐。

温馨提示

对已有key的索引会在后台异步进行。 您可通过FT.INFO跟踪进度。在回填完成前，搜索结果可能不完整。
Search模块实现了精简的FT.*命令集。 针对向量搜索，已覆盖所有核心功能：HNSW与精确KNN、混合过滤及FT.AGGREGATE。

完整参考请见Valkey search官方文档。