RediSearch 向量数据库完整指南

Redis 本身是一个高性能的键值数据库，但它也在向量检索领域逐步增强了自己的能力，特别是在引入 [Redis-Search（RediSearch）模块] 后，支持了向量相似度检索（Vector Similarity Search），从而具备了初步的向量数据库功能。

✅ Redis 向量数据库

✅ Redis 向量数据库的核心概念

1. 向量检索基本原理

向量数据库的核心任务是：在大量向量中，找出与给定查询向量最相似的若干个向量（TopK）。常用的度量方式：

• L2 距离（欧式距离）

• Inner Product（内积 / 点积）

• Cosine Similarity（余弦相似度）

Redis 借助 RediSearch 模块 实现向量检索能力，支持上面这些主流的距离度量。

2. Redis 向量索引机制

Redis 的向量索引是通过 FT.CREATE 命令建立的。索引类型支持：

• FLAT：暴力搜索，慢但精确。

• HNSW：近似搜索，速度快，适合大规模向量场景。

你可以定义一个 schema，其中某个字段是向量，例如：

FT.CREATE my_index ON HASH PREFIX 1 doc: SCHEMA vec_field VECTOR HNSW 6 TYPE FLOAT32 DIM 128 DISTANCE_METRIC COSINE ...

🚀 如何快速搭建 Redis 向量数据库（Docker + RediSearch）

1. 启动 Redis 向量数据库（Docker）

docker run -d --name redis-vector \
  -p 6379:6379 \
  redis/redis-stack-server:latest

这个镜像包含了 Redis + RediSearch（向量支持）等模块。

你可以通过 redis-cli 或 Python 等客户端连接：

docker exec -it redis-vector redis-cli

2. 创建向量索引（以 128 维为例）

FT.CREATE vec_index ON HASH PREFIX 1 doc: SCHEMA vec VECTOR HNSW 6 TYPE FLOAT32 DIM 128 DISTANCE_METRIC COSINE

3. 插入向量数据（假设用 base64 编码的二进制向量）

HSET doc:1 vec "\x00\x00\x80\x3f\x00\x00\x00\x40..."   # 128维 float32 向量

注意：Redis 使用 二进制格式的 float32 向量，必须按顺序序列化为字节数组（Python 可用 struct.pack）。

4. 查询最相似的向量（TopK）

FT.SEARCH vec_index "*=>[KNN 3 @vec $BLOB]" PARAMS 2 BLOB "\x00\x00\x80\x3f..." DIALECT 2

返回结果为 Top 3 最相似的文档及其距离分数。

🧪 Python 使用示例（推荐：redis-py + numpy）

import redis  
import numpy as np  
  
# 配置参数  
REDIS_HOST = 'localhost'  
REDIS_PORT = 6379  
VECTOR_DIM = 128  
INDEX_NAME = 'vec_index'  
DOC_PREFIX = 'doc:'  
  
# 建立连接  
r = redis.Redis(host=REDIS_HOST, port=REDIS_PORT)  
  
# 删除旧索引（如果存在）  
try:  
    r.execute_command("FT.DROPINDEX", INDEX_NAME, "DD")  
except redis.ResponseError:  
    pass  # 索引不存在则跳过  
  
VECTOR_DIM = 128  
  
create_cmd = [  
    "FT.CREATE", INDEX_NAME, "ON", "HASH", "PREFIX", "1", DOC_PREFIX,  
    "SCHEMA", "vec", "VECTOR", "HNSW", "12",   # 注意这里是 12 个参数  
    "TYPE", "FLOAT32",  
    "DIM", str(VECTOR_DIM),  
    "DISTANCE_METRIC", "COSINE",  
    "INITIAL_CAP", "1000",  
    "M", "16",  
    "EF_CONSTRUCTION", "200"  
]  
r.execute_command(*create_cmd)  
  
  
  
# 插入 N 个随机向量  
N = 5  
for i in range(N):  
    vec = np.random.rand(VECTOR_DIM).astype(np.float32)  
    key = f"{DOC_PREFIX}{i}"  
    r.hset(key, mapping={"vec": vec.tobytes()})  
print(f"✅ 成功插入 {N} 个向量")  
  
# 构造查询向量  
query_vec = np.random.rand(VECTOR_DIM).astype(np.float32)  
query_blob = query_vec.tobytes()  
  
# 执行 KNN 检索（Top 3）  
res = r.execute_command(  
    "FT.SEARCH", INDEX_NAME,  
    "*=>[KNN 3 @vec $BLOB]",  
    "PARAMS", "2", "BLOB", query_blob,  
    "DIALECT", "2"  
)  
  
# 解析查询结果为结构化数据  
def parse_knn_result(raw):  
    count = raw[0]  
    results = []  
    for i in range(1, len(raw), 2):  
        doc_id = raw[i].decode()  
        fields = raw[i + 1]  
        field_map = {fields[j].decode(): fields[j + 1] for j in range(0, len(fields), 2)}  
        vector = np.frombuffer(field_map["vec"], dtype=np.float32)  
        score = float(field_map.get("__vec_score", 0))  
        results.append({"doc_id": doc_id, "score": score, "vector": vector})  
    return results  
  
parsed_results = parse_knn_result(res)  
  
# 输出结果  
print("\n🔍 KNN 查询结果（Top 3）：")  
for i, item in enumerate(parsed_results):  
    print(f"{i+1}. 文档ID: {item['doc_id']}, 相似度分数: {item['score']:.6f}")  
  
# 可选：比较原始向量与查询向量的余弦相似度  
def cosine_sim(a, b):  
    return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))  
  
print("\n🧪 余弦相似度验证（与查询向量）:")  
for item in parsed_results:  
    sim = cosine_sim(query_vec, item["vector"])  
    print(f"{item['doc_id']}: {sim:.6f}")

✅ Redis 向量数据库优劣对比总结

🧭 建议

• 适合中小规模、混合 KV + 向量场景（如：推荐系统、图文搜索）

• 若追求极致性能/大规模向量，建议考虑 Faiss、Milvus、Qdrant 等专业向量数据库

✅ 向量检索的文本推荐系统 Demo 实现

✅ 1. 文本预处理与向量化

• 使用 TfidfVectorizer 将原始文本转换为 高维稀疏向量（TF-IDF）。

• 适合小规模场景，但对语义捕捉能力有限（可替换为 BERT、FastText 等语义模型）。

✅ 2. 量化向量

• 将原始 float32 向量 线性量化为 uint8 类型，压缩存储和提升检索效率。

• 线性量化方式：

  q=⌊x−xmin⁡step⌋q = \left\lfloor \frac{x - x_{\min}}{\text{step}} \right\rfloor

  其中 step 为动态计算的离散分辨率。

✅ 3. Redis 存储与索引（简化版）

• 使用 Redis 的 HSET 命令将量化向量作为字节流存储到 Hash。

• 同时创建了一个 RediSearch 向量索引（HNSW），便于未来扩展原生向量搜索。

• 当前查询仍使用 Python 本地计算（Redis 向量检索可作为下一步优化）。

✅ 4. 查询流程

• 对用户输入进行相同的向量化 + 量化。

• 从 Redis 获取所有存储向量，计算与查询向量的 余弦相似度。

• 对相似度进行降序排序，输出 Top-N 推荐结果。

✅ 5. 优点 & 局限

✅ 适合的使用场景

• 教学、原型验证、Demo 展示

• 小规模文本推荐（几千条以内）

• 对资源约束较强的系统（量化节省空间）

pip install scikit-learn

import numpy as np  
import redis  
from sklearn.feature_extraction.text import TfidfVectorizer  
from sklearn.metrics.pairwise import cosine_similarity  
  
# 示例文本数据  
documents = [  
    "I love programming in Python",  
    "Python is great for data science",  
    "Machine learning is a subset of artificial intelligence",  
    "Deep learning can be used for image recognition"  
]  
  
# 创建 TF-IDF 向量器  
vectorizer = TfidfVectorizer()  
tfidf_matrix = vectorizer.fit_transform(documents).toarray()  
  
# 量化函数（8-bit）  
def quantize_vector(vec, num_bits=8):  
    min_val = np.min(vec)  
    max_val = np.max(vec)  
    if max_val == min_val:  
        return np.zeros_like(vec, dtype=np.uint8), min_val, 1e-5  
    step = (max_val - min_val) / (2**num_bits - 1)  
    qvec = np.round((vec - min_val) / step).astype(np.uint8)  
    return qvec, min_val, step  
  
# 量化所有文本向量  
quantized_vectors = []  
for vec in tfidf_matrix:  
    qvec, _, _ = quantize_vector(vec)  
    quantized_vectors.append(qvec)  
  
# Redis连接  
r = redis.Redis(host='localhost', port=6379)  
  
# 初始化 RediSearch 索引（已存在则忽略）  
try:  
    r.execute_command(  
        'FT.CREATE', 'text_index',  
        'ON', 'HASH',  
        'PREFIX', '1', 'doc:',  
        'SCHEMA', 'vec', 'VECTOR', 'HNSW', '6',  
        'TYPE', 'FLOAT32',  
        'DIM', str(len(tfidf_matrix[0])),  
        'DISTANCE_METRIC', 'COSINE'  
    )  
except redis.ResponseError as e:  
    if "Index already exists" in str(e):  
        pass  
    else:  
        raise e  
  
# 存储向量到 Redisfor idx, qvec in enumerate(quantized_vectors):  
    r.hset(f"doc:{idx+1}", "vec", qvec.tobytes())  
  
# 查询文本  
query = "I love learning data science"  
  
# TF-IDF 向量化 + 量化  
query_vec = vectorizer.transform([query]).toarray()[0]  
quantized_query_vec, _, _ = quantize_vector(query_vec)  
  
# 从 Redis 读取所有量化向量  
stored_vectors = []  
for i in range(len(documents)):  
    vec_bytes = r.hget(f"doc:{i+1}", "vec")  
    if vec_bytes:  
        stored_vectors.append(np.frombuffer(vec_bytes, dtype=np.uint8))  
  
# 计算余弦相似度  
similarities = cosine_similarity([quantized_query_vec], stored_vectors)[0]  
  
# 结果排序并返回 Top Ntop_n = 3  
top_indices = np.argsort(similarities)[::-1][:top_n]  
  
print("Top 推荐结果：")  
for idx in top_indices:  
    print(f"- 文本: {documents[idx]}（相似度: {similarities[idx]:.4f}）")

下面是一个使用 MiniLM 进行向量化，并将向量存储到 Redis 原生向量索引（HNSW + RediSearch） 中的完整推荐系统代码。

✅ 整体架构流程：

1. 文本向量化：使用 sentence-transformers 中的 MiniLM 模型。

2. 存储向量到 Redis：使用 RediSearch 的 VECTOR 类型 + HNSW 索引。

3. 查询向量相似文本：通过 RediSearch FT.SEARCH 执行 ANN 检索，直接在 Redis 中完成。

🔧 先决条件

pip install redis sentence-transformers numpy

启动支持 RediSearch 和 VECTOR 的 Redis（用官方 docker 镜像）：

docker run -d --name redis-vec -p 6379:6379 redis/redis-stack:latest

✅ 完整代码实现

import redis  
import numpy as np  
from sentence_transformers import SentenceTransformer  
  
# 初始化 MiniLM 模型  
model = SentenceTransformer('all-MiniLM-L6-v2')  
  
# 示例文本数据  
documents = [  
    "I love programming in Python",  
    "Python is great for data science",  
    "Machine learning is a subset of artificial intelligence",  
    "Deep learning can be used for image recognition"  
]  
  
# 向量化文本  
vectors = model.encode(documents).astype(np.float32)  
  
# Redis 连接  
r = redis.Redis(host='localhost', port=6379)  
  
# 删除旧索引（若已存在）  
try:  
    r.execute_command("FT.DROPINDEX", "text_index", "DD")  
except redis.ResponseError:  
    pass  
  
# 创建 RediSearch 向量索引（HNSW）  
r.execute_command(  
    "FT.CREATE", "text_index",  
    "ON", "HASH",  
    "PREFIX", "1", "doc:",  
    "SCHEMA",  
    "content", "TEXT",  
    "vec", "VECTOR", "HNSW", "14",  
    "TYPE", "FLOAT32",  
    "DIM", "384",  
    "DISTANCE_METRIC", "COSINE",  
    "INITIAL_CAP", "1000",  
    "M", "16",  
    "EF_CONSTRUCTION", "200",  
    "EF_RUNTIME", "100"  
)  
  
# 存储向量和内容  
for idx, (vec, content) in enumerate(zip(vectors, documents)):  
    r.hset(  
        f"doc:{idx + 1}",  
        mapping={  
            "content": content,  
            "vec": vec.tobytes()  
        }  
    )  
  
# 查询文本  
query = "What is useful for machine learning and AI?"  
query_vec = model.encode([query])[0].astype(np.float32).tobytes()  
  
  
print("\n🔍 查询文本:" + query)  
  
# 执行向量搜索 - 简化版本  
print("\n🔍 执行向量搜索:")  
  
# 接收整个结果集  
res = r.execute_command(  
    "FT.SEARCH", "text_index",  
    "*=>[KNN 3 @vec $vec_param]",  
    "PARAMS", "2", "vec_param", query_vec,  
    "RETURN", "1", "content",  # 只返回内容字段  
    "DIALECT", "2"  
)  
  
# 获取结果总数  
total_results = res[0]  
print(f"找到 {total_results} 个匹配结果")  
  
# 一种更稳健的解析方法  
documents = []  
for i in range(1, len(res), 2):  
    doc_id = res[i]  
    doc_id_str = doc_id.decode('utf-8') if isinstance(doc_id, bytes) else str(doc_id)  
  
    # 获取文档属性  
    doc_attrs = res[i + 1]  
  
    # 初始化一个文档字典  
    doc_dict = {"id": doc_id_str}  
  
    # 解析文档属性 (扁平列表转换为字典)  
    for j in range(0, len(doc_attrs), 2):  
        if j + 1 < len(doc_attrs):  
            key = doc_attrs[j]  
            key_str = key.decode('utf-8') if isinstance(key, bytes) else str(key)  
  
            value = doc_attrs[j + 1]  
            if isinstance(value, bytes):  
                try:  
                    value_str = value.decode('utf-8')  
                except UnicodeDecodeError:  
                    value_str = str(value)  
            else:  
                value_str = str(value)  
  
            doc_dict[key_str] = value_str  
  
    documents.append(doc_dict)  
  
# 打印结果  
for i, doc in enumerate(documents):  
    print(f"\n{i + 1}. 文档ID: {doc['id']}")  
  
    if 'content' in doc:  
        print(f"   内容: {doc['content']}")  
    else:  
        print("   内容: 未找到")  
  
# 执行另一个查询 - 使用SORTBY获取分数  
print("\n\n执行另一个查询 - 使用SORTBY获取分数:")  
  
try:  
    # 使用 SORTBY DESC 获取最相似的结果优先显示  
    res2 = r.execute_command(  
        "FT.SEARCH", "text_index",  
        "*=>[KNN 3 @vec $vec_param AS score]",  
        "PARAMS", "2", "vec_param", query_vec,  
        "RETURN", "2", "content", "score",  
        "SORTBY", "score", "DESC",  # 添加 DESC 参数  
        "DIALECT", "2"  
    )  
  
    total_results = res2[0]  
    print(f"找到 {total_results} 个匹配结果")  
  
    # 解析结果  
    documents = []  
    for i in range(1, len(res2), 2):  
        doc_id = res2[i]  
        doc_id_str = doc_id.decode('utf-8') if isinstance(doc_id, bytes) else str(doc_id)  
  
        # 获取文档属性  
        doc_attrs = res2[i + 1]  
  
        # 初始化一个文档字典  
        doc_dict = {"id": doc_id_str}  
  
        # 解析文档属性 (扁平列表转换为字典)  
        for j in range(0, len(doc_attrs), 2):  
            if j + 1 < len(doc_attrs):  
                key = doc_attrs[j]  
                key_str = key.decode('utf-8') if isinstance(key, bytes) else str(key)  
  
                value = doc_attrs[j + 1]  
                if isinstance(value, bytes):  
                    try:  
                        value_str = value.decode('utf-8')  
                    except UnicodeDecodeError:  
                        value_str = str(value)  
                else:  
                    value_str = str(value)  
  
                doc_dict[key_str] = value_str  
  
        documents.append(doc_dict)  
  
    # 打印结果  
    for i, doc in enumerate(documents):  
        print(f"\n{i + 1}. 文档ID: {doc['id']}")  
  
        if 'content' in doc:  
            print(f"   内容: {doc['content']}")  
        if 'score' in doc:  
            try:  
                score = float(doc['score'])  
                print(f"   相似度得分: {score:.4f}")  
            except ValueError:  
                print(f"   相似度得分: {doc['score']} (转换为浮点数失败)")  
        else:  
            print("   相似度得分: 未找到")  
  
except Exception as e:  
    print(f"第二次查询出错: {str(e)}")  
  
# 尝试第三种方法 - 使用简单的HGETALL来获取数据  
print("\n\n尝试第三种方法 - 简单查询所有文档:")  
  
try:  
    # 首先获取向量搜索结果中的文档ID  
    res3 = r.execute_command(  
        "FT.SEARCH", "text_index",  
        "*=>[KNN 3 @vec $vec_param AS score]",  
        "PARAMS", "2", "vec_param", query_vec,  
        "NOCONTENT",  # 只返回文档ID  
        "DIALECT", "2"  
    )  
  
    total_results = res3[0]  
    print(f"找到 {total_results} 个文档ID")  
  
    # 收集文档ID  
    doc_ids = []  
    for i in range(1, len(res3)):  
        doc_id = res3[i]  
        doc_id_str = doc_id.decode('utf-8') if isinstance(doc_id, bytes) else str(doc_id)  
        doc_ids.append(doc_id_str)  
  
    # 对每个文档ID使用HGETALL获取完整内容  
    for i, doc_id in enumerate(doc_ids):  
        doc_data = r.hgetall(doc_id)  
        print(f"\n{i + 1}. 文档ID: {doc_id}")  
  
        for key, value in doc_data.items():  
            key_str = key.decode('utf-8') if isinstance(key, bytes) else str(key)  
  
            # 内容字段解码为文本  
            if key_str == 'content' and isinstance(value, bytes):  
                value_str = value.decode('utf-8')  
            # 向量字段保持为bytes  
            elif key_str == 'vec' and isinstance(value, bytes):  
                value_str = f"<二进制数据: {len(value)} 字节>"  
            # 其他字段  
            else:  
                try:  
                    value_str = value.decode('utf-8') if isinstance(value, bytes) else str(value)  
                except UnicodeDecodeError:  
                    value_str = str(value)  
  
            print(f"   {key_str}: {value_str}")  
  
except Exception as e:  
    print(f"第三次查询出错: {str(e)}")

✅ 示例输出

python script.py

🔍 查询文本:What is useful for machine learning and AI?

🔍 执行向量搜索:
找到 3 个匹配结果

1. 文档ID: doc:2
   内容: Python is great for data science

2. 文档ID: doc:3
   内容: Machine learning is a subset of artificial intelligence

3. 文档ID: doc:4
   内容: Deep learning can be used for image recognition

执行另一个查询 - 使用SORTBY获取分数:
找到 3 个匹配结果

1. 文档ID: doc:2
   内容: Python is great for data science
   相似度得分: 0.6823

2. 文档ID: doc:4
   内容: Deep learning can be used for image recognition
   相似度得分: 0.5540

3. 文档ID: doc:3
   内容: Machine learning is a subset of artificial intelligence
   相似度得分: 0.3589

尝试第三种方法 - 简单查询所有文档:
找到 3 个文档ID

1. 文档ID: doc:2
   vec: <二进制数据: 1536 字节>
   content: Python is great for data science

2. 文档ID: doc:3
   vec: <二进制数据: 1536 字节>
   content: Machine learning is a subset of artificial intelligence

3. 文档ID: doc:4
   vec: <二进制数据: 1536 字节>
   content: Deep learning can be used for image recognition

进程已结束，退出代码为 0

✅ 总结与优势

相关文章

Redis 技术栈

• Redis 内存数据库 - Redis 基础配置与使用
• Redis 主从复制异常 - 高可用架构
• RedisMQ 消息队列 - 基于 Redis 的轻量级消息队列

向量数据库与搜索

• Elasticsearch 搜索引擎 - 全文搜索解决方案

性能优化

• Redis 性能调优 - 优化策略
• 索引设计最佳实践 - 索引优化
• 向量搜索优化 - ANN 算法调优