向量数据库存储与检索（RAG向）

向量数据库

它是专门把文本、图像、音频等非结构化数据转成高维向量后，再进行“语义级”存储、检索与管理的专用数据库。

核心概念

词向量 / 嵌入（Embedding）：将单词、句子甚至整篇文章映射成固定维度的浮点向量（如 1536 维）。语义相近的内容在向量空间里距离更近。

向量数据库：以「向量」为第一等公民的数据库系统，支持相似度计算（余弦、点积、欧氏距离）和高维索引（HNSW、IVF、PQ 等。

与传统数据库的区别

常用向量数据库

向量检索

核心概念

给定一个「查询向量」，在庞大的向量集合里毫秒级找出与其最相似的 Top-K 个向量，并返回对应原始对象（文本、图片、商品等）的过程。

应用场景

• RAG：把用户问题向量与知识库向量匹配，召回最相关段落。
• 推荐：用用户行为向量找相似商品/视频。
• 以图搜图：图片 → 向量 → 向量检索。
• 异常检测：与“正常”向量距离过远即异常。

代码实现

我们使用相对比较熟悉的Redis来演示对于向量的一些重要操作。

Redis是一个开源的缓存数据库，拥有集群功能稳定，访问速度快等非常多特性，通常用于存储键值对数据。部署Redis时需要注意，默认开源的Redis社区版本是不支持向量数据存储的，需要额外安装redissearch模块，才能支持向量数据存储。

这里一个大家一种比较简单的部署方式，就是使用docker部署一个带有redissearch模块的redis容器。docker部署指令为

docker run --name redis -p 6379:6379 -d redis/redis-stack:latest
# 安装redis库
pip install redis
pip install redisearch  

代码如下：

import os
import dotenv
import dashscope
import redis
import numpy as np
from http import HTTPStatus
from redis.commands.search.field import TextField, VectorField
from redis.commands.search.index_definition import IndexDefinition
from redis.commands.search.query import Query

# ========== 配置 ==========
# 加载 .env 文件中的环境变量
dotenv.load_dotenv()
# 设置通义千问 API Key
dashscope.api_key = os.getenv("DASHSCOPE_API_KEY")

# 定义 Redis 向量索引名称
INDEX_NAME = "embedding_index"
# 设置向量维度，需与所使用的 embedding 模型输出维度一致
VECTOR_DIM = 1024
# 设置向量相似度计算方式为余弦距离
DISTANCE_METRIC = "COSINE"

# ========== 连接 Redis ==========
# 初始化 Redis 客户端连接
# 注意：为了正确存储二进制向量数据，关闭了响应解码功能
redis_client = redis.Redis(
    host="localhost",
    port=6379,
    password=None,
    decode_responses=False  # 存向量要关掉 decode
)

# ========== 创建索引（只执行一次） ==========
def create_index():
    """
    创建 Redis 向量搜索索引。
    
    如果索引已存在则跳过创建，否则根据预定义的字段结构创建新索引。
    索引包括文本字段和向量字段，使用 HNSW 算法进行向量近似最近邻搜索。
    """
    try:
        # 尝试获取索引信息以判断是否已存在
        redis_client.ft(INDEX_NAME).info()
        print("✅ 索引已存在")
    except Exception:  # 统一捕获异常
        # 创建新的向量索引
        redis_client.ft(INDEX_NAME).create_index(
            [
                TextField("text"),  # 文本字段用于存储原始文本
                VectorField(
                    "embedding",  # 向量字段名
                    "HNSW",       # 使用 HNSW 算法
                    {"TYPE": "FLOAT32", "DIM": VECTOR_DIM, "DISTANCE_METRIC": DISTANCE_METRIC}
                )
            ],
            definition=IndexDefinition(prefix=["doc:"])  # 建议加上前缀
        )
        print("✅ 已创建向量索引")

# ========== 写入一条数据 ==========
def insert_text(text: str):
    """
    调用通义千问 embedding 接口并将文本及其向量表示写入 Redis。
    
    参数:
        text (str): 需要转换为向量并存储的原始文本内容。
    """
    # 调用多模态 embedding 接口获取文本向量
    resp = dashscope.MultiModalEmbedding.call(
        model="multimodal-embedding-v1",
        input=[{"text": text}]
    )

    if resp.status_code == HTTPStatus.OK:
        # 提取 embedding 向量并转换为字节格式
        embedding = resp.output["embeddings"][0]["embedding"]
        vector = np.array(embedding, dtype=np.float32).tobytes()
        # 构造 Redis 键名
        key = f"doc:{resp.request_id}"
        # 将文本和向量写入 Redis Hash 结构中
        redis_client.hset(key, mapping={
            "text": text,
            "embedding": vector
        })
        print(f"✅ 已写入 Redis，key={key}, 向量维度={len(embedding)}")
    else:
        print(f"❌ 调用失败: {resp.code}, {resp.message}")

# ========== 相似度搜索 ==========
def search_similar(query_text: str, topk: int = 1):
    """
    根据输入文本查询与其最相似的文本列表。
    
    参数:
        query_text (str): 查询用的文本内容。
        topk (int): 返回最相似结果的数量，默认为 1。
    """
    # 获取查询文本的 embedding 向量
    resp = dashscope.MultiModalEmbedding.call(
        model="multimodal-embedding-v1",
        input=[{"text": query_text}]
    )

    if resp.status_code != HTTPStatus.OK:
        print(f"❌ 查询 embedding 失败: {resp.code}, {resp.message}")
        return

    # 将查询向量转换为字节格式
    query_vector = np.array(
        resp.output["embeddings"][0]["embedding"], dtype=np.float32
    ).tobytes()

    # 构造 KNN 查询语句
    knn_query = f'*=>[KNN {topk} @embedding $vec_param]'
    q = Query(knn_query).sort_by("__embedding_score").paging(0, topk)

    # 执行向量相似性搜索
    search_result = redis_client.ft(INDEX_NAME).search(
        q, query_params={"vec_param": query_vector}
    )

    print(f"🔍 与 '{query_text}' 最相似的 {topk} 条：")
    # 输出匹配结果
    for i, doc in enumerate(search_result.docs, 1):
        print(f"{i}. {doc.text}")

# ========== 使用示例 ==========
if __name__ == "__main__":
    # 创建索引
    create_index()
    # 插入示例数据
    insert_text("我喜欢吃苹果")
    insert_text("苹果是我最喜欢吃的水果")
    insert_text("我喜欢用苹果手机")
    # 相似度搜索
    search_similar("我喜欢用小米")

执行结果

✅ 已创建向量索引
✅ 已写入 Redis，key=doc:e86bd66c-295a-92fe-9174-c799a38da841, 向量维度=1024
✅ 已写入 Redis，key=doc:6c3b9196-7d85-9d6e-95f4-a0eddde938bd, 向量维度=1024
✅ 已写入 Redis，key=doc:73ce1dad-cedb-95b8-bd5e-b5b2c4cd3f63, 向量维度=1024
🔍 与 '我喜欢用小米' 最相似的 1 条：
1. 我喜欢用苹果手机

查看数据库

-c799a38da841, 向量维度=1024
✅ 已写入 Redis，key=doc:6c3b9196-7d85-9d6e-95f4-a0eddde938bd, 向量维度=1024
✅ 已写入 Redis，key=doc:73ce1dad-cedb-95b8-bd5e-b5b2c4cd3f63, 向量维度=1024
🔍 与 ‘我喜欢用小米’ 最相似的 1 条：

1. 我喜欢用苹果手机

查看数据库

![在这里插入图片描述](https://i-blog.csdnimg.cn/direct/c4f6afe0834f43c79929bd427b7af982.jpeg#pic_center)