使用 Redis 作为向量数据库

news2025/6/2 8:37:03

一、什么是向量数据库？

向量（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 redis

client = 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 requests

URL = ("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 SentenceTransformer

embedder = 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 JSON
  PREFIX 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 Query

knn_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 pd

def 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()}) \
            .docs
        for 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))