RedisVectorSearch：向量相似度搜索在推荐与问答中的应用

好的，没问题，直接进入主题：

各位朋友，大家好！今天咱们聊聊Redis Vector Search，这玩意儿听起来高大上，其实说白了就是让Redis这个老伙计学会了“看脸”，哦不，是“看向量”，然后帮你找长得像的东西。这技术在推荐系统和问答系统中可是大有用武之地。

第一章：Redis Vector Search是个啥？

Redis大家应该都熟，一个高性能的键值数据库，速度快得飞起。但是以前的Redis，只会存字符串、数字、列表啥的，都是些“死板”的数据。现在好了，Redis有了Vector Search，它能存向量了！

啥是向量？简单说，就是一堆数字，用来表示一个东西的特征。比如，一篇文章的向量，可以表示这篇文章的主题、风格等等。两篇文章的向量越接近，就说明它们越相似。

Redis Vector Search就是让你把这些向量存到Redis里，然后它可以帮你快速地找到和某个向量最相似的向量。

第二章：为啥要用Redis Vector Search？

可能有朋友会问，向量搜索的技术多了去了，为啥要用Redis Vector Search？

• 快！ 这是Redis的看家本领，向量搜索也继承了这个优点。基于内存的索引，搜起来那叫一个行云流水。
• 简单！ Redis用起来很方便，API简单易懂，上手快。
• 省钱！ 如果你已经用了Redis，那直接用Vector Search，不用再引入新的数据库，省钱！
• 集成方便！ Redis生态系统完善，可以很容易地和其他组件集成。

当然，Redis Vector Search也有缺点：

• 内存限制： 所有数据都存在内存里，所以数据量不能太大。
• 功能相对简单： 相比于专业的向量数据库，Redis Vector Search的功能还不够丰富。

第三章：Redis Vector Search的核心概念

要玩转Redis Vector Search，得先了解几个核心概念：

• Index（索引）： 就像书的目录一样，Redis Vector Search用索引来加速搜索。
• Vector Field（向量字段）： 存储向量的字段。
• Similarity Metric（相似度度量）： 用来衡量两个向量相似度的算法。常用的有：
  • COSINE (余弦相似度): 最常用，计算两个向量夹角的余弦值，值越大越相似。
  • L2 (欧几里得距离): 计算两个向量之间的距离，距离越小越相似。
  • IP (内积): 直接计算两个向量的内积，值越大越相似（向量需要归一化）。

第四章：Redis Vector Search实战：推荐系统

咱们来用Redis Vector Search做一个简单的电影推荐系统。

1. 准备数据：

假设我们有一些电影的数据，每部电影都有一个标题和一个描述。我们需要把电影的描述转换成向量。可以用一些现成的模型，比如BERT、Word2Vec等等。这里我们假设已经有了电影向量数据，格式如下：

movies = [
    {"id": "1", "title": "复仇者联盟", "vector": [0.1, 0.2, 0.3, 0.4, 0.5]},
    {"id": "2", "title": "钢铁侠", "vector": [0.2, 0.3, 0.4, 0.5, 0.6]},
    {"id": "3", "title": "美国队长", "vector": [0.3, 0.4, 0.5, 0.6, 0.7]},
    {"id": "4", "title": "雷神", "vector": [0.4, 0.5, 0.6, 0.7, 0.8]},
    {"id": "5", "title": "黑豹", "vector": [0.5, 0.6, 0.7, 0.8, 0.9]},
]

2. 连接Redis：

import redis

# 连接Redis
r = redis.Redis(host='localhost', port=6379, db=0)

3. 创建索引：

# 定义索引名称
index_name = "movie_index"

# 定义向量字段名称
vector_field_name = "movie_vector"

# 定义索引结构
schema = (
    redis.commands.search.SchemaField.Vector(vector_field_name, "FLAT", {"TYPE": "FLOAT32", "DIM": 5, "DISTANCE_METRIC": "COSINE"}),
    redis.commands.search.SchemaField.Text("title") # 添加title字段方便查看结果
)

# 创建索引
try:
    r.ft(index_name).create_index(fields=schema)
except redis.exceptions.ResponseError as e:
    if str(e) == "Index already exists":
        print("Index already exists")
    else:
        raise e

这里我们创建了一个名为movie_index的索引，向量字段名为movie_vector，使用FLAT索引（暴力搜索，数据量小的时候效果好），向量维度是5，相似度度量方式是COSINE。 FLAT索引适用于数据量较小的情况，它会遍历所有向量进行比较。 对于大规模数据，可以使用更高效的索引算法，比如HNSW。

4. 导入数据：

# 导入电影数据
for movie in movies:
    movie_id = movie["id"]
    movie_title = movie["title"]
    movie_vector = movie["vector"]

    # 将向量转换为字节数组
    movie_vector_bytes = bytes(np.array(movie_vector).astype(np.float32))

    # 使用HSET存储数据
    r.hset(f"movie:{movie_id}", mapping={
        vector_field_name: movie_vector_bytes,
        "title": movie_title  # 存储电影标题
    })

这里我们使用HSET命令将电影数据存储到Redis中，key是movie:{movie_id}，field是movie_vector和title，value分别是电影向量和电影标题。 注意，需要将向量转换为字节数组再存储。

5. 进行搜索：

import numpy as np

# 假设用户喜欢复仇者联盟
user_vector = movies[0]["vector"]

# 将用户向量转换为字节数组
user_vector_bytes = bytes(np.array(user_vector).astype(np.float32))

# 构建查询语句
query = f"*=>[KNN 3 @{vector_field_name} $user_vector AS score]"
params = {"user_vector": user_vector_bytes}

# 执行查询
results = r.ft(index_name).search(query, query_params=params)

# 输出结果
for doc in results.docs:
    print(f"Movie ID: {doc.id}, Title: {doc.title}, Score: {doc.score}")

这里我们假设用户喜欢复仇者联盟，所以我们用复仇者联盟的向量作为用户向量。然后我们构建了一个查询语句，使用KNN算法找到和用户向量最相似的3部电影。 KNN 3表示返回最相似的3个结果。 @movie_vector指定在movie_vector字段上进行向量搜索。 $user_vector AS score 将相似度得分存储在名为score的字段中。

完整的代码示例：

import redis
import numpy as np

# 连接Redis
r = redis.Redis(host='localhost', port=6379, db=0)

# 定义电影数据
movies = [
    {"id": "1", "title": "复仇者联盟", "vector": [0.1, 0.2, 0.3, 0.4, 0.5]},
    {"id": "2", "title": "钢铁侠", "vector": [0.2, 0.3, 0.4, 0.5, 0.6]},
    {"id": "3", "title": "美国队长", "vector": [0.3, 0.4, 0.5, 0.6, 0.7]},
    {"id": "4", "title": "雷神", "vector": [0.4, 0.5, 0.6, 0.7, 0.8]},
    {"id": "5", "title": "黑豹", "vector": [0.5, 0.6, 0.7, 0.8, 0.9]},
]

# 定义索引名称
index_name = "movie_index"

# 定义向量字段名称
vector_field_name = "movie_vector"

# 定义索引结构
schema = (
    redis.commands.search.SchemaField.Vector(vector_field_name, "FLAT", {"TYPE": "FLOAT32", "DIM": 5, "DISTANCE_METRIC": "COSINE"}),
    redis.commands.search.SchemaField.Text("title")
)

# 创建索引
try:
    r.ft(index_name).create_index(fields=schema)
except redis.exceptions.ResponseError as e:
    if str(e) == "Index already exists":
        print("Index already exists")
    else:
        raise e

# 导入电影数据
for movie in movies:
    movie_id = movie["id"]
    movie_title = movie["title"]
    movie_vector = movie["vector"]

    # 将向量转换为字节数组
    movie_vector_bytes = bytes(np.array(movie_vector).astype(np.float32))

    # 使用HSET存储数据
    r.hset(f"movie:{movie_id}", mapping={
        vector_field_name: movie_vector_bytes,
        "title": movie_title
    })

# 假设用户喜欢复仇者联盟
user_vector = movies[0]["vector"]

# 将用户向量转换为字节数组
user_vector_bytes = bytes(np.array(user_vector).astype(np.float32))

# 构建查询语句
query = f"*=>[KNN 3 @{vector_field_name} $user_vector AS score]"
params = {"user_vector": user_vector_bytes}

# 执行查询
results = r.ft(index_name).search(query, query_params=params)

# 输出结果
for doc in results.docs:
    print(f"Movie ID: {doc.id}, Title: {doc.title}, Score: {doc.score}")

第五章：Redis Vector Search实战：问答系统

咱们再来用Redis Vector Search做一个简单的问答系统。

1. 准备数据：

假设我们有一些问题和答案的数据，我们需要把问题转换成向量。可以用一些现成的模型，比如Sentence Transformers等等。这里我们假设已经有了问题向量数据，格式如下：

questions = [
    {"id": "1", "question": "什么是人工智能？", "vector": [0.1, 0.2, 0.3, 0.4, 0.5]},
    {"id": "2", "question": "人工智能有哪些应用？", "vector": [0.2, 0.3, 0.4, 0.5, 0.6]},
    {"id": "3", "question": "机器学习是什么？", "vector": [0.3, 0.4, 0.5, 0.6, 0.7]},
    {"id": "4", "question": "深度学习是什么？", "vector": [0.4, 0.5, 0.6, 0.7, 0.8]},
    {"id": "5", "question": "自然语言处理是什么？", "vector": [0.5, 0.6, 0.7, 0.8, 0.9]},
]

answers = {
    "1": "人工智能是研究、开发用于模拟、延伸和扩展人的智能的理论、方法、技术及应用系统的一门新的技术科学。",
    "2": "人工智能的应用非常广泛，包括机器学习、自然语言处理、计算机视觉等等。",
    "3": "机器学习是一种实现人工智能的方法，它使计算机能够从数据中学习，而无需进行明确的编程。",
    "4": "深度学习是一种机器学习技术，它使用多层神经网络来分析数据。",
    "5": "自然语言处理是一种人工智能技术，它使计算机能够理解和处理人类语言。"
}

2. 连接Redis：

import redis

# 连接Redis
r = redis.Redis(host='localhost', port=6379, db=0)

3. 创建索引：

# 定义索引名称
index_name = "question_index"

# 定义向量字段名称
vector_field_name = "question_vector"

# 定义索引结构
schema = (
    redis.commands.search.SchemaField.Vector(vector_field_name, "FLAT", {"TYPE": "FLOAT32", "DIM": 5, "DISTANCE_METRIC": "COSINE"}),
    redis.commands.search.SchemaField.Text("question")
)

# 创建索引
try:
    r.ft(index_name).create_index(fields=schema)
except redis.exceptions.ResponseError as e:
    if str(e) == "Index already exists":
        print("Index already exists")
    else:
        raise e

4. 导入数据：

import numpy as np

# 导入问题数据
for question in questions:
    question_id = question["id"]
    question_text = question["question"]
    question_vector = question["vector"]

    # 将向量转换为字节数组
    question_vector_bytes = bytes(np.array(question_vector).astype(np.float32))

    # 使用HSET存储数据
    r.hset(f"question:{question_id}", mapping={
        vector_field_name: question_vector_bytes,
        "question": question_text
    })

5. 进行搜索：

import numpy as np

# 用户提问
user_question = "人工智能有哪些应用场景？"

# 将用户问题转换为向量 (这里需要用和之前训练问题向量相同的模型)
user_vector = [0.25, 0.35, 0.45, 0.55, 0.65]  # 假设转换后的向量

# 将用户向量转换为字节数组
user_vector_bytes = bytes(np.array(user_vector).astype(np.float32))

# 构建查询语句
query = f"*=>[KNN 1 @{vector_field_name} $user_vector AS score]"
params = {"user_vector": user_vector_bytes}

# 执行查询
results = r.ft(index_name).search(query, query_params=params)

# 输出结果
for doc in results.docs:
    question_id = doc.id.split(":")[1]
    answer = answers[question_id]
    print(f"问题: {doc.question}")
    print(f"答案: {answer}")

完整的代码示例：

import redis
import numpy as np

# 连接Redis
r = redis.Redis(host='localhost', port=6379, db=0)

# 定义问题数据
questions = [
    {"id": "1", "question": "什么是人工智能？", "vector": [0.1, 0.2, 0.3, 0.4, 0.5]},
    {"id": "2", "question": "人工智能有哪些应用？", "vector": [0.2, 0.3, 0.4, 0.5, 0.6]},
    {"id": "3", "question": "机器学习是什么？", "vector": [0.3, 0.4, 0.5, 0.6, 0.7]},
    {"id": "4", "question": "深度学习是什么？", "vector": [0.4, 0.5, 0.6, 0.7, 0.8]},
    {"id": "5", "question": "自然语言处理是什么？", "vector": [0.5, 0.6, 0.7, 0.8, 0.9]},
]

# 定义答案数据
answers = {
    "1": "人工智能是研究、开发用于模拟、延伸和扩展人的智能的理论、方法、技术及应用系统的一门新的技术科学。",
    "2": "人工智能的应用非常广泛，包括机器学习、自然语言处理、计算机视觉等等。",
    "3": "机器学习是一种实现人工智能的方法，它使计算机能够从数据中学习，而无需进行明确的编程。",
    "4": "深度学习是一种机器学习技术，它使用多层神经网络来分析数据。",
    "5": "自然语言处理是一种人工智能技术，它使计算机能够理解和处理人类语言。"
}

# 定义索引名称
index_name = "question_index"

# 定义向量字段名称
vector_field_name = "question_vector"

# 定义索引结构
schema = (
    redis.commands.search.SchemaField.Vector(vector_field_name, "FLAT", {"TYPE": "FLOAT32", "DIM": 5, "DISTANCE_METRIC": "COSINE"}),
    redis.commands.search.SchemaField.Text("question")
)

# 创建索引
try:
    r.ft(index_name).create_index(fields=schema)
except redis.exceptions.ResponseError as e:
    if str(e) == "Index already exists":
        print("Index already exists")
    else:
        raise e

# 导入问题数据
for question in questions:
    question_id = question["id"]
    question_text = question["question"]
    question_vector = question["vector"]

    # 将向量转换为字节数组
    question_vector_bytes = bytes(np.array(question_vector).astype(np.float32))

    # 使用HSET存储数据
    r.hset(f"question:{question_id}", mapping={
        vector_field_name: question_vector_bytes,
        "question": question_text
    })

# 用户提问
user_question = "人工智能有哪些应用场景？"

# 将用户问题转换为向量 (这里需要用和之前训练问题向量相同的模型)
user_vector = [0.25, 0.35, 0.45, 0.55, 0.65]  # 假设转换后的向量

# 将用户向量转换为字节数组
user_vector_bytes = bytes(np.array(user_vector).astype(np.float32))

# 构建查询语句
query = f"*=>[KNN 1 @{vector_field_name} $user_vector AS score]"
params = {"user_vector": user_vector_bytes}

# 执行查询
results = r.ft(index_name).search(query, query_params=params)

# 输出结果
for doc in results.docs:
    question_id = doc.id.split(":")[1]
    answer = answers[question_id]
    print(f"问题: {doc.question}")
    print(f"答案: {answer}")

第六章：Redis Vector Search的进阶技巧

• HNSW索引： 对于大规模数据，可以使用HNSW索引，它是一种近似最近邻搜索算法，速度更快，但精度略有下降。
• 向量归一化： 在使用内积作为相似度度量方式时，需要对向量进行归一化，保证向量的模长为1。
• 混合查询： 可以将向量搜索和其他类型的查询（比如文本搜索、范围查询）结合起来，实现更复杂的搜索需求。
• 数据更新： Redis Vector Search支持数据的实时更新，可以动态地添加、删除和修改向量。

第七章：总结

Redis Vector Search是一个简单、高效的向量搜索解决方案，特别适合于对性能要求高，数据量不大的场景。它可以应用于推荐系统、问答系统、图像搜索等领域。当然，它也有一些局限性，比如内存限制、功能不够丰富等等。在实际应用中，需要根据具体的需求选择合适的解决方案。

第八章：一些补充说明

• 关于向量化： 上面两个例子中，我们都假设已经有了向量数据。但在实际应用中，需要自己将文本、图像等数据转换成向量。可以使用一些现成的模型，比如BERT、Word2Vec、Sentence Transformers等等。
• 关于模型选择： 选择合适的向量化模型非常重要，它直接影响搜索的准确性。需要根据具体的业务场景选择合适的模型。
• 关于参数调优： Redis Vector Search有很多参数可以调整，比如索引类型、相似度度量方式、KNN算法的参数等等。需要根据实际情况进行调优，才能达到最佳的性能。
• 关于数据规模： Redis Vector Search适合于数据量不大的场景。如果数据量太大，可以考虑使用专业的向量数据库，比如Milvus、Weaviate等等。

第九章：问题与解答

欢迎大家提问，我会尽力解答。

表格总结：

特性	Redis Vector Search	专业向量数据库 (例如 Milvus, Weaviate)
优点	速度快，简单易用，集成方便，成本较低	功能丰富，支持大规模数据，索引类型多样
缺点	内存限制，功能相对简单	部署和维护成本较高，学习曲线较陡峭
适用场景	小规模数据，对性能要求高，已使用Redis的场景	大规模数据，需要复杂的功能，对性能要求极高的场景
索引类型	FLAT, HNSW	HNSW, IVF, ANNOY 等更多选择
相似度度量	COSINE, L2, IP	更多选择
数据更新	支持实时更新	支持实时更新
混合查询	支持 (例如：文本 + 向量)	支持 (通常更加灵活)
存储介质	内存	内存 + 磁盘 (通常支持持久化)
社区与生态系统	Redis 社区	专业向量数据库社区

希望今天的讲座对大家有所帮助！ 谢谢大家！