本快速入门指南可帮助您：
- 了解什么是向量数据库
- 创建 Redis 向量数据库
- 创建向量嵌入并存储向量
- 查询数据并执行矢量搜索
了解向量数据库
数据通常是非结构化的，这意味着它没有由定义良好的架构描述。非结构化数据的示例包括文本段落、图像、视频或音频。存储和搜索非结构化数据的一种方法是使用向量嵌入。
什么是向量？在机器学习和人工智能中，向量是表示数据的数字序列。它们是模型的输入和输出，以数字形式封装基础信息。矢量将非结构化数据（如文本、图像、视频和音频）转换为机器学习模型可以处理的格式。
为什么它们很重要？向量捕获数据中固有的复杂模式和语义含义，使其成为各种应用程序的强大工具。它们允许机器学习模型更有效地理解和操作非结构化数据。
增强传统搜索。传统的关键字或词汇搜索依赖于单词或短语的精确匹配，这可能会受到限制。相比之下，向量搜索或语义搜索利用了向量嵌入中捕获的丰富信息。通过将数据映射到向量空间中，相似的项目根据其含义彼此靠近放置。这种方法允许更准确和有意义的搜索结果，因为它考虑了查询的上下文和语义内容，而不仅仅是使用的确切单词。
创建 Redis 向量数据库
您可以将 Redis Stack 用作向量数据库。它允许您：
- 将向量和关联的元数据存储在哈希或 JSON 文档中
- 创建和配置用于搜索的二级索引
- 执行矢量搜索
- 更新矢量和元数据
- 删除和清理
安装所需的 Python 包
创建一个 Python 虚拟环境，并使用以下命令安装 pip 以下依赖项：
- redis ：您可以在本文档站点的客户端部分找到有关客户端库的 redis-py 更多详细信息。
- pandas ：Pandas 是一个数据分析库。
- sentence-transformers ：您将使用 SentenceTransformers 框架在全文上生成嵌入。
- tabulate ： pandas 用于 tabulate 渲染 Markdown。
您还需要在 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 ：
client = redis.Redis(host="localhost", port=6379, decode_responses=True)
准备演示数据集
本快速入门指南还使用了自行车数据集。下面是其中的示例文档：
{ "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 字段包含自行车的自由格式文本描述，将用于创建矢量嵌入。
1. 获取演示数据
您需要首先将演示数据集作为 JSON 数组获取：
URL = ("https://raw.githubusercontent.com/bsbodden/redis_vss_getting_started" "/main/data/bikes.json" ) response = requests.get(URL, timeout=10) bikes = response.json()
检查其中一个自行车 JSON 文档的结构：
json.dumps(bikes[0], indent=2)
2. 将演示数据存储在 Redis 中
现在，使用 JSON 遍历 bikes 数组，将数据作为 JSON 文档存储在 Redis 中。SET 命令。以下代码使用管道来最大程度地减少网络往返时间：
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 文档中检索特定属性：
res = client.json().get("bikes:010", "$.model") # >>> ['Summit']
3. 选择文本嵌入模型
HuggingFace 有大量的文本嵌入模型目录，这些模型可通过 SentenceTransformers 框架在本地提供服务。在这里，我们使用广泛用于搜索引擎、聊天机器人和其他 AI 应用程序的 MS MARCO 模型。
from sentence_transformers import SentenceTransformer embedder = SentenceTransformer('msmarco-distilbert-base-v4')
4. 生成文本嵌入
遍历所有 Redis 键，前缀 bikes: 为：
keys = sorted(client.keys("bikes:*")) # >>> ['bikes:001', 'bikes:002', ..., 'bikes:011']
使用键作为 JSON 的输入。MGET 命令以及 $.description 字段，用于收集列表中的描述。然后，将说明列表传递给 .encode() 该方法：
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 将矢量化描述插入到 Redis 中的自行车文档中。SET 命令。以下命令将一个新字段插入到 JSONPath 下的每个文档 $.description_embeddings 中。再一次，使用管道执行此操作，以避免不必要的网络往返：
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 命令：
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 数组。在上面的示例中，为了便于阅读，数组被大大缩短。
创建索引
1. 创建带有向量字段的索引
必须创建索引才能查询文档元数据或执行矢量搜索。使用 FT.CREATE 命令：
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
以下是 VECTOR 字段定义的细分：
- $.description_embeddings AS vector ：向量字段的 JSON 路径及其字段别名 vector 。
- FLAT ：指定索引方法，该方法可以是平面索引，也可以是分层可导航的小世界图 （HNSW）。
- TYPE FLOAT32 ：设置向量分量的浮点精度，在本例中为 32 位浮点数。
- DIM 768 ：嵌入的长度或尺寸，由所选的嵌入模型确定。
- DISTANCE_METRIC COSINE ：所选距离函数：余弦距离。
您可以在矢量参考文档中找到有关所有这些选项的更多详细信息。
2. 检查索引的状态
一旦您执行 FT.CREATE 命令，索引过程在后台运行。在很短的时间内，所有 JSON 文档都应该被索引并准备好进行查询。若要验证这一点，可以使用 FT.INFO 命令，该命令提供有关索引的详细信息和统计信息。特别值得一提的是成功编制索引的文档数量和失败的文档数量：
FT.INFO idx:bikes_vss
执行矢量搜索
本快速入门指南重点介绍矢量搜索。但是，您可以在文档数据库快速入门指南中了解有关如何基于文档元数据进行查询的详细信息。
1. 嵌入您的查询
以下代码片段显示了将用于在 Redis 中执行矢量搜索的文本查询列表：
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 模型将每个输入查询编码为向量嵌入：
encoded_queries = embedder.encode(queries) len(encoded_queries) # >>> 11
使用与嵌入文档相同的嵌入模型来嵌入查询至关重要。使用不同的模型将导致语义搜索结果不佳或错误。
2. K-最近邻 （KNN） 搜索
KNN 算法根据所选的距离函数计算查询向量与 Redis 中每个向量之间的距离。然后，它返回与查询向量距离最小的前 K 项。这些是语义上最相似的项目。
现在构造一个查询来做到这一点：
query = ( Query('(*)=>[KNN 3 @vector $query_vector AS vector_score]') .sort_by('vector_score') .return_fields('vector_score', 'id', 'brand', 'model', 'description') .dialect(2) )
让我们分解上面的查询模板：
- filter 表达式的 (*) 意思是 all 。换言之，未应用任何过滤。您可以将其替换为按其他元数据进行筛选的表达式。
- 查询 KNN 的部分搜索前 3 个最近邻。
- 查询向量必须作为参数传入 query_vector 。
- 到查询向量的距离返回为 vector_score 。
- 结果按以下 vector_score 方式排序。
- 最后，它返回每个结果的字段 vector_score 、 id 、 brand 和 model description 。
若要在 FT.SEARCH 命令中使用向量查询，必须指定 DIALECT 2 或更高版本。
您必须将矢量化查询作为参数名称 query_vector 为 的字节数组传递。以下代码从查询向量创建一个 Python NumPy 数组，并将其转换为可作为参数传递给查询的紧凑的字节级表示形式：
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 表以可视化结果：
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、品牌和型号。
例如，对于查询“最适合儿童的山地自行车”，相似度得分最高 （ 0.54 ），因此最接近的匹配项是“Nord”品牌的“Chook air 5”自行车型号，描述为：
Chook Air 5 为 6 岁及以上的孩子提供了一辆耐用且超轻便的山地自行车，让他们第一次体验赛道，并在森林和田野中轻松巡航。下部顶部管便于在任何情况下安装和拆卸，让您的孩子在小径上更加安全。Chook Air 5 是山地自行车的完美入门。
从描述来看，这款自行车非常适合年幼的孩子，并且嵌入准确地捕捉了描述的语义。
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...
后续步骤
- 您可以通过阅读矢量参考文档来了解有关查询选项（如筛选器和矢量范围查询）的更多信息。
- 完整的搜索和查询文档可能会让您感兴趣。
- 如果要以交互方式遵循代码示例，则可以使用启发本快速入门指南的 Jupyter 笔记本。
- 如果您想查看 Redis 向量数据库的更多高级示例，请访问 GitHub 上的 Redis AI 资源页面。
