Redis 作为矢量数据库快速入门指南

了解如何将 Redis 用作矢量数据库

本快速入门指南可帮助您：

了解什么是矢量数据库
创建 Redis 矢量数据库
创建向量嵌入并存储向量
查询数据并执行向量搜索

了解矢量数据库

数据通常是非结构化的，这意味着它不是由定义明确的架构描述的。非结构化数据的示例包括文本段落、图像、视频或音频。存储和搜索非结构化数据的一种方法是使用向量嵌入。

**什么是向量？**在机器学习和 AI 中，向量是表示数据的数字序列。它们是模型的输入和输出，以数字形式封装基础信息。向量将非结构化数据（如文本、图像、视频和音频）转换为机器学习模型可以处理的格式。

**为什么它们很重要？**向量捕获数据中固有的复杂模式和语义含义，使其成为适用于各种应用的强大工具。它们使机器学习模型能够更有效地理解和操作非结构化数据。
**增强传统搜索。**传统的关键字或词法搜索依赖于单词或短语的精确匹配，这可能会有限制。相比之下，向量搜索或语义搜索利用向量嵌入中捕获的丰富信息。通过将数据映射到向量空间，相似的项目会根据它们的含义彼此靠近。这种方法允许更准确和有意义的搜索结果，因为它考虑查询的上下文和语义内容，而不仅仅是使用的确切单词。

创建 Redis 矢量数据库

您可以将 Redis Stack 用作矢量数据库。它允许您：

将向量和关联的元数据存储在哈希或 JSON 文档中
创建和配置用于搜索的二级索引
执行向量搜索
更新向量和元数据
删除和清理

最简单的入门方法是使用 Redis Cloud：

创建一个免费帐户。
按照说明创建免费数据库

这个免费的 Redis Cloud 数据库开箱即用，具有所有 Redis Stack 功能。

您也可以使用安装指南在本地计算机上安装 Redis Stack。

您需要为 Redis 服务器配置以下功能：JSON 和 Search and query。

安装所需的 Python 软件包

创建 Python 虚拟环境并使用pip安装以下依赖项：

redis：您可以在本文档站点的 clients 部分找到有关redis-py客户端库的更多详细信息。
pandas：Pandas 是一个数据分析库。
sentence-transformers：您将使用 SentenceTransformers 框架在全文上生成嵌入。
tabulate：pandas用tabulate呈现 Markdown。

您还需要在 Python 代码中导入以下内容：

python

import json
import time

import numpy as np
import pandas as pd
import requests
import redis
from redis.commands.search.field import (
    NumericField,
    TagField,
    TextField,
    VectorField,
)
from redis.commands.search.indexDefinition import IndexDefinition, IndexType
from redis.commands.search.query import Query
from sentence_transformers import SentenceTransformer

连接

连接到 Redis。默认情况下，Redis 返回二进制响应。要解码它们，请将参数decode_responses设置为True ：

python

client = redis.Redis(host="localhost", port=6379, decode_responses=True)

提示

您可以从 Redis Cloud 数据库配置页面复制并粘贴连接详细信息，而不是使用本地 Redis Stack 服务器。以下是托管在 AWS 区域us-east-1中并侦听端口 16379 的云数据库的示例连接字符串：redis-16379.c283.us-east-1-4.ec2.cloud.redislabs.com:16379。连接字符串的格式为host:port .您还必须复制并粘贴 Cloud 数据库的用户名和密码。用于连接默认用户的代码行将更改为client = redis.Redis(host="redis-16379.c283.us-east-1-4.ec2.cloud.redislabs.com", port=16379, password="your_password_here", decode_responses=True) 。

准备 demo 数据集

本快速入门指南还使用 bikes 数据集。下面是其中的示例文档：

json

{
  "model": "Jigger",
  "brand": "Velorim",
  "price": 270,
  "type": "Kids bikes",
  "specs": {
    "material": "aluminium",
    "weight": "10"
  },
  "description": "Small and powerful, the Jigger is the best ride for the smallest of tikes! ..."
}

字段description 包含自行车的自由格式文本描述，将用于创建矢量嵌入。

获取 demo 数据

您需要首先以 JSON 数组的形式获取演示数据集：

python

URL = ("https://raw.githubusercontent.com/bsbodden/redis_vss_getting_started"
       "/main/data/bikes.json"
       )
response = requests.get(URL, timeout=10)
bikes = response.json()

检查其中一个 bike JSON 文档的结构：

python

json.dumps(bikes[0], indent=2)

将 demo 数据存储在 Redis 中

现在迭代数组bikes，使用 JSON.SET命令将数据作为 JSON 文档存储在 Redis 中。以下代码使用pipeline来最小化网络往返时间：

python

pipeline = client.pipeline()
for i, bike in enumerate(bikes, start=1):
    redis_key = f"bikes:{i:03}"
    pipeline.json().set(redis_key, "$", bike)
res = pipeline.execute()
# >>> [True, True, True, True, True, True, True, True, True, True, True]

加载后，您可以使用 JSONPath 表达式从 Redis 中的某个 JSON 文档中检索特定属性：

python

res = client.json().get("bikes:010", "$.model")
# >>> ['Summit']

选择文本嵌入模型

HuggingFace 有一个庞大的文本嵌入模型目录，这些模型可通过SentenceTransformers框架在本地提供服务。在这里，我们使用了广泛用于搜索引擎、聊天机器人和其他 AI 应用程序的 MS MARCO 模型。

python

from sentence_transformers import SentenceTransformer

embedder = SentenceTransformer('msmarco-distilbert-base-v4')

生成文本嵌入

迭代所有带有前缀bikes:的 Redis 键：

python

keys = sorted(client.keys("bikes:*"))
# >>> ['bikes:001', 'bikes:002', ..., 'bikes:011']

使用keys和$.description属性作为JSON.MGET命令的输入。以及字段，以将描述收集到一个列表中。然后，将descriptions列表传递给.encode()方法：

python

descriptions = client.json().mget(keys, "$.description")
descriptions = [item for sublist in descriptions for item in sublist]
embedder = SentenceTransformer("msmarco-distilbert-base-v4")
embeddings = embedder.encode(descriptions).astype(np.float32).tolist()
VECTOR_DIMENSION = len(embeddings[0])
# >>> 768

使用 JSON.SET命令将矢量化描述插入到 Redis 中的 bike 文档中。以下命令将新字段插入到 JSONPath 下的$.description_embeddings每个文档中。同样，使用管道执行此操作，以避免不必要的网络往返：

python

pipeline = client.pipeline()
for key, embedding in zip(keys, embeddings):
    pipeline.json().set(key, "$.description_embeddings", embedding)
pipeline.execute()
# >>> [True, True, True, True, True, True, True, True, True, True, True]

使用 JSON.GET 命令检查更新的 bike 文档之一：

python

res = client.json().get("bikes:010")
# >>>
# {
#   "model": "Summit",
#   "brand": "nHill",
#   "price": 1200,
#   "type": "Mountain Bike",
#   "specs": {
#     "material": "alloy",
#     "weight": "11.3"
#   },
#   "description": "This budget mountain bike from nHill performs well..."
#   "description_embeddings": [
#     -0.538114607334137,
#     -0.49465855956077576,
#     -0.025176964700222015,
#     ...
#   ]
# }

提示

在 JSON 文档中存储向量嵌入时，嵌入将存储为 JSON 数组。在上面的示例中，为了可读性，数组被大大缩短了。

创建索引

创建具有 vector 字段的索引

您必须创建索引才能查询文档元数据或执行矢量搜索。使用 FT.CREATE 命令：

Redis CLIPython

bash

FT.CREATE idx:bikes_vss ON JSON
  PREFIX 1 bikes: SCORE 1.0
  SCHEMA
    $.model TEXT WEIGHT 1.0 NOSTEM
    $.brand TEXT WEIGHT 1.0 NOSTEM
    $.price NUMERIC
    $.type TAG SEPARATOR ","
    $.description AS description TEXT WEIGHT 1.0
    $.description_embeddings AS vector VECTOR FLAT 6 TYPE FLOAT32 DIM 768 DISTANCE_METRIC COSINE

python

schema = (
    TextField("$.model", no_stem=True, as_name="model"),
    TextField("$.brand", no_stem=True, as_name="brand"),
    NumericField("$.price", as_name="price"),
    TagField("$.type", as_name="type"),
    TextField("$.description", as_name="description"),
    VectorField(
        "$.description_embeddings",
        "FLAT",
        {
            "TYPE": "FLOAT32",
            "DIM": VECTOR_DIMENSION,
            "DISTANCE_METRIC": "COSINE",
        },
        as_name="vector",
    ),
)
definition = IndexDefinition(prefix=["bikes:"], index_type=IndexType.JSON)
res = client.ft("idx:bikes_vss").create_index(fields=schema, definition=definition)
# >>> 'OK'

以下是字段VECTOR定义的细分：

$.description_embeddings AS vector：向量字段vector的 JSON 路径及其字段别名。
FLAT：指定索引方法，可以是平面索引或分层可导航小世界图（HNSW）。
TYPE FLOAT32：设置矢量分量的浮点精度，在本例中为 32 位浮点数。
DIM 768：嵌入的长度或尺寸，由所选嵌入模型确定
DISTANCE_METRIC COSINE：选择的距离函数：余弦距离。

您可以在矢量参考文档中找到有关所有这些选项的更多详细信息。

检查索引的状态

一旦你执行 FT.CREATE 命令，则索引过程将在后台运行。在短时间内，所有 JSON 文档都应该被索引并准备好进行查询。要验证这一点，您可以使用 FT.INFO 命令，该命令提供有关索引的详细信息和统计信息。特别值得关注的是成功索引的文档数和失败数：

Redis CLIPython

bash

FT.INFO idx:bikes_vss

python

info = client.ft("idx:bikes_vss").info()
num_docs = info["num_docs"]
indexing_failures = info["hash_indexing_failures"]
# print(f"{num_docs} documents indexed with {indexing_failures} failures")
# >>> 11 documents indexed with 0 failures

执行向量搜索

本快速入门指南重点介绍向量搜索。但是，您可以在文档数据库快速入门指南中了解有关如何基于文档元数据进行查询的更多信息。

嵌入查询

以下代码片段显示了您将用于在 Redis 中执行矢量搜索的文本查询列表：

python

queries = [
    "Bike for small kids",
    "Best Mountain bikes for kids",
    "Cheap Mountain bike for kids",
    "Female specific mountain bike",
    "Road bike for beginners",
    "Commuter bike for people over 60",
    "Comfortable commuter bike",
    "Good bike for college students",
    "Mountain bike for beginners",
    "Vintage bike",
    "Comfortable city bike",
]

首先，使用相同的 SentenceTransformers 模型将每个输入查询编码为向量嵌入：

python

encoded_queries = embedder.encode(queries)
len(encoded_queries)
# >>> 11

提示

请务必使用与文档相同的嵌入模型来嵌入查询。使用不同的模型将导致语义搜索结果不佳或错误。

K 最近邻（KNN）搜索

KNN 算法根据选择的距离函数计算查询向量与 Redis 中每个向量之间的距离。然后，它返回与查询向量距离最小的前 K 个项目。这些是语义上最相似的项目。

现在构造一个查询来执行此操作：

python

query = (
    Query('(*)=>[KNN 3 @vector $query_vector AS vector_score]')
     .sort_by('vector_score')
     .return_fields('vector_score', 'id', 'brand', 'model', 'description')
     .dialect(2)
)

让我们分解上面的查询模板：

筛选条件表达式(*)表示所有，换句话说，未应用筛选。您可以将其替换为按其他元数据筛选的表达式
查询部分KNN搜索前 3 个最近邻域。
查询向量必须作为参数query_vector 传入。
到查询向量的距离返回vector_score
结果按 vector_score 排序。
最后，它为每个结果返回字段 vector_score, id, brand, model, 和 description

提示

要使用 FT.SEARCH 命令，则必须指定 DIALECT 2 或更高版本。

您必须将矢量化查询作为参数 query_vector 的字节数组传递。以下代码从查询向量创建一个 Python NumPy 数组，并将其转换为可以作为参数传递给查询的紧凑的字节级表示形式：

python

client.ft('idx:bikes_vss').search(
    query,
    {
      'query_vector': np.array(encoded_query, dtype=np.float32).tobytes()
    }
).docs

有了查询的模板，您就可以在一个循环中执行所有查询。请注意，脚本将为每一个结果计算vector_score 作为1 - doc.vector_score 。由于余弦距离用作度量，因此距离最小的项目更近，因此与查询更相似。

然后，遍历匹配的文档并创建一个结果列表，这些结果可以转换为 Pandas 表以可视化结果：

python

def create_query_table(query, queries, encoded_queries, extra_params=None):
    """
    Creates a query table.
    """
    results_list = []
    for i, encoded_query in enumerate(encoded_queries):
        result_docs = (
            client.ft("idx:bikes_vss")
            .search(
                query,
                {"query_vector": np.array(encoded_query, dtype=np.float32).tobytes()}
                | (extra_params if extra_params else {}),
            )
            .docs
        )
        for doc in result_docs:
            vector_score = round(1 - float(doc.vector_score), 2)
            results_list.append(
                {
                    "query": queries[i],
                    "score": vector_score,
                    "id": doc.id,
                    "brand": doc.brand,
                    "model": doc.model,
                    "description": doc.description,
                }
            )

    # Optional: convert the table to Markdown using Pandas
    queries_table = pd.DataFrame(results_list)
    queries_table.sort_values(
        by=["query", "score"], ascending=[True, False], inplace=True
    )
    queries_table["query"] = queries_table.groupby("query")["query"].transform(
        lambda x: [x.iloc[0]] + [""] * (len(x) - 1)
    )
    queries_table["description"] = queries_table["description"].apply(
        lambda x: (x[:497] + "...") if len(x) > 500 else x
    )
    return queries_table.to_markdown(index=False)

查询结果显示各个查询的前三个匹配项（我们的 K 参数）以及每个查询的自行车 ID、品牌和型号。

例如，对于查询“Best Mountain bikes for kids”，相似度得分最高（0.54），因此最接近的匹配是“Nord”品牌的“Chook air 5”自行车型号，描述如下：

"The Chook Air 5 gives kids aged six years and older a durable and uberlight mountain bike for their first experience on tracks and easy cruising through forests and fields. The lower top tube makes it easy to mount and dismount in any situation, giving your kids greater safety on the trails. The Chook Air 5 is the perfect intro to mountain biking."

从描述来看，这辆自行车非常适合年幼的孩子，嵌入准确地捕捉了描述的语义。

python

query = (
    Query("(*)=>[KNN 3 @vector $query_vector AS vector_score]")
    .sort_by("vector_score")
    .return_fields("vector_score", "id", "brand", "model", "description")
    .dialect(2)
)

table = create_query_table(query, queries, encoded_queries)
print(table)
# >>> | Best Mountain bikes for kids     |    0.54 | bikes:003...

后续步骤

您可以通过阅读向量参考文档来了解有关查询选项的更多信息，例如过滤器和向量范围查询。
您可能对完整的 Redis 查询引擎文档感兴趣。
如果您想以更具交互性的方式遵循代码示例，则可以使用启发本快速入门指南的 Jupyter Notebook。
如果您想查看 Redis 矢量数据库的更多高级示例，请访问 GitHub 上的 Redis AI 资源页面。

Redis 作为矢量数据库快速入门指南 ​

了解矢量数据库 ​

创建 Redis 矢量数据库 ​

安装所需的 Python 软件包 ​

连接 ​

准备 demo 数据集 ​

创建索引 ​

执行向量搜索 ​

后续步骤 ​

Redis 作为矢量数据库快速入门指南

了解矢量数据库

创建 Redis 矢量数据库

安装所需的 Python 软件包

连接

准备 demo 数据集

创建索引

执行向量搜索

后续步骤