使用 Redis 作为向量数据库

一、什么是向量数据库？

向量（Vector）：在机器学习和 AI 中，向量是由一系列数字组成的序列，用于数值化地描述数据的特征或语义。文本、图像、音频等非结构化数据可以通过模型转换成固定长度的向量。
向量数据库：专门存储、索引和检索向量的数据库系统。可以基于向量之间的距离度量（如余弦相似度、欧氏距离等）进行高效的近邻搜索（Nearest Neighbor Search），从而实现“语义搜索”或“相似度搜索”。
与传统搜索的区别：
- 传统搜索依赖于关键词精确匹配，无法识别同义词、上下文或语义抽象。
- 向量搜索通过将数据空间映射到高维向量空间，使语义相近的内容在向量空间中距离更近，从而返回更符合用户意图的结果。

二、准备工作

本文示例使用 Python 客户端库 RedisVL，以及常见的 Python 生态组件：

# 建议在虚拟环境中安装
pip install redis pandas sentence-transformers tabulate redisvl

说明

redis：官方 Python 客户端。
pandas：用于结果展示。
sentence-transformers：生成文本向量。
tabulate：渲染 Markdown 表格。
redisvl：Redis 向量搜索专用扩展（可选，本文使用原生 redis.commands.search API）。

三、连接 Redis

如果你使用本地 Redis：

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

如果使用 Redis Cloud，则将 host、port、password 替换为云端实例参数：

client = redis.Redis(host="redis-16379.c283.us-east-1-4.ec2.cloud.redislabs.com",port=16379,password="your_password_here",decode_responses=True,
)

四、准备示例数据集

本文使用开源的 bikes 数据集，每条记录包含如下字段：

{"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! ..."
}

1. 拉取数据

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

2. 存储到 Redis（JSON 文档）

pipeline = client.pipeline()
for i, bike in enumerate(bikes, start=1):key = f"bikes:{i:03}"pipeline.json().set(key, "$", bike)
pipeline.execute()

你可以这样读取某个字段：

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

五、生成并存储向量嵌入

1. 选择文本嵌入模型

from sentence_transformers import SentenceTransformerembedder = SentenceTransformer('msmarco-distilbert-base-v4')

2. 批量获取描述并生成向量

import numpy as np# 获取所有 key
keys = sorted(client.keys("bikes:*"))# 批量读取 description
descs = client.json().mget(keys, "$.description")
# 扁平化列表
descriptions = [item for sublist in descs for item in sublist]# 生成嵌入并转换为 float32 列表
embeddings = embedder.encode(descriptions).astype(np.float32).tolist()
VECTOR_DIM = len(embeddings[0])  # 768

3. 插入向量字段

pipeline = client.pipeline()
for key, vec in zip(keys, embeddings):pipeline.json().set(key, "$.description_embeddings", vec)
pipeline.execute()

此时，每条记录都多了一个 $.description_embeddings 数组字段。

六、创建检索索引

为了同时支持基于字段和基于向量的搜索，需要创建一个 Redis Search 索引：

# 在 Redis CLI 环境中执行
FT.CREATE idx:bikes_vss ON JSONPREFIX 1 bikes:SCHEMA$.model                TEXT    WEIGHT 1.0 NOSTEM$.brand                TEXT    WEIGHT 1.0 NOSTEM$.price                NUMERIC$.type                 TAG     SEPARATOR ","$.description          TEXT    WEIGHT 1.0$.description_embeddings AS vector VECTOR FLAT \TYPE FLOAT32 DIM 768 DISTANCE_METRIC COSINE

FLAT：扁平索引；也可使用 HNSW（图索引）以提高速度与扩展性。
TYPE FLOAT32：32 位浮点。
DIM 768：向量维度。
DISTANCE_METRIC COSINE：余弦相似度。

创建完成后，通过 FT.INFO idx:bikes_vss 可以查看索引状态，确认文档是否全部就绪。

七、执行向量搜索

1. 嵌入查询文本

queries = ["Bike for small kids","Best Mountain bikes for kids","Cheap Mountain bike for kids",# ... 共 11 条
]
encoded_queries = embedder.encode(queries)

注意：必须使用与文档相同的模型和参数，否则语义相似度会大打折扣。

2. 构造 KNN 查询模板

from redis.commands.search.query import Queryknn_query = (Query("(*)=>[KNN 3 @vector $qvector AS score]").sort_by("score").return_fields("score", "id", "brand", "model", "description").dialect(2)
)

(*)：不过滤，检索全集。
KNN 3：返回最相近的 3 个向量。
@vector $qvector：向量字段名与占位符。
dialect(2)：必要参数以支持向量查询语法。

3. 执行查询并展示

import pandas as pddef run_search(queries, encoded_qs):rows = []for q, vec in zip(queries, encoded_qs):docs = client.ft("idx:bikes_vss") \.search(knn_query, {"qvector": np.array(vec, dtype=np.float32).tobytes()}) \.docsfor doc in docs:rows.append({"query": q,"score": round(1 - float(doc.score), 2),"id":    doc.id,"brand": doc.brand,"model": doc.model,"desc":  doc.description[:100] + "..."})df = pd.DataFrame(rows)return df.sort_values(["query","score"], ascending=[True,False])table = run_search(queries, encoded_queries)
print(table.to_markdown(index=False))