向量数据库存储与检索RAG向向量数据库它是专门把文本、图像、音频等非结构化数据转成高维向量后再进行“语义级”存储、检索与管理的专用数据库。核心概念词向量 / 嵌入Embedding将单词、句子甚至整篇文章映射成固定维度的浮点向量如 1536 维。语义相近的内容在向量空间里距离更近。向量数据库以「向量」为第一等公民的数据库系统支持相似度计算余弦、点积、欧氏距离和高维索引HNSW、IVF、PQ 等。与传统数据库的区别特性关系型数据库 (MySQL…)词向量数据库存储单元行/列高维稠密向量查询语言SQL向量相似度检索k-NN索引结构B 树HNSW、IVF、PQ、DiskANN适用场景精确匹配、事务语义搜索、推荐、RAG、多媒体检索常用向量数据库向量数据库数据存储方式索引类型优势特点使用场景部署形式Redis Stack内存磁盘HNSW、Flat高速读写、低延迟、支持 KV 和向量混合存储、内置向量检索实时推荐、在线问答、搜索增强单机/集群/DockerMilvus磁盘内存IVF、HNSW、ANNOY专业向量数据库支持大规模向量检索、高并发、分布式多模态检索、AI 搜索、图片/视频相似度单机/分布式Weaviate文档 向量HNSW开箱即用的语义搜索支持 GraphQL、向量 文本属性混合查询企业知识库、语义搜索、聊天机器人单机/CloudPinecone云托管HNSW、IVFSaaS 服务自动扩展管理简单无运维压力AI 应用向量搜索、推荐系统CloudQdrant内存 磁盘HNSW高性能、支持 REST/gRPC容易与 Python 集成文本/图像向量检索、AI 应用单机/集群/DockerFAISS内存为主IVF、PQ、HNSW高性能向量索引库GPU 加速适合大规模离线检索大规模向量计算、科研实验、批量搜索Python/C 库向量检索核心概念给定一个「查询向量」在庞大的向量集合里毫秒级找出与其最相似的 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( hostlocalhost, port6379, passwordNone, decode_responsesFalse # 存向量要关掉 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} ) ], definitionIndexDefinition(prefix[doc:]) # 建议加上前缀 ) print(✅ 已创建向量索引) # 写入一条数据 def insert_text(text: str): 调用通义千问 embedding 接口并将文本及其向量表示写入 Redis。 参数: text (str): 需要转换为向量并存储的原始文本内容。 # 调用多模态 embedding 接口获取文本向量 resp dashscope.MultiModalEmbedding.call( modelmultimodal-embedding-v1, input[{text: text}] ) if resp.status_code HTTPStatus.OK: # 提取 embedding 向量并转换为字节格式 embedding resp.output[embeddings][0][embedding] vector np.array(embedding, dtypenp.float32).tobytes() # 构造 Redis 键名 key fdoc:{resp.request_id} # 将文本和向量写入 Redis Hash 结构中 redis_client.hset(key, mapping{ text: text, embedding: vector }) print(f✅ 已写入 Rediskey{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( modelmultimodal-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], dtypenp.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(我喜欢用小米)执行结果✅ 已创建向量索引 ✅ 已写入 Rediskeydoc:e86bd66c-295a-92fe-9174-c799a38da841, 向量维度1024 ✅ 已写入 Rediskeydoc:6c3b9196-7d85-9d6e-95f4-a0eddde938bd, 向量维度1024 ✅ 已写入 Rediskeydoc:73ce1dad-cedb-95b8-bd5e-b5b2c4cd3f63, 向量维度1024 与 我喜欢用小米 最相似的 1 条 1. 我喜欢用苹果手机查看数据库-c799a38da841, 向量维度1024✅ 已写入 Rediskeydoc:6c3b9196-7d85-9d6e-95f4-a0eddde938bd, 向量维度1024✅ 已写入 Rediskeydoc:73ce1dad-cedb-95b8-bd5e-b5b2c4cd3f63, 向量维度1024 与 ‘我喜欢用小米’ 最相似的 1 条我喜欢用苹果手机查看数据库 
向量数据库存储与检索
发布时间：2026/7/16 2:21:32
向量数据库存储与检索RAG向向量数据库它是专门把文本、图像、音频等非结构化数据转成高维向量后再进行“语义级”存储、检索与管理的专用数据库。核心概念词向量 / 嵌入Embedding将单词、句子甚至整篇文章映射成固定维度的浮点向量如 1536 维。语义相近的内容在向量空间里距离更近。向量数据库以「向量」为第一等公民的数据库系统支持相似度计算余弦、点积、欧氏距离和高维索引HNSW、IVF、PQ 等。与传统数据库的区别特性关系型数据库 (MySQL…)词向量数据库存储单元行/列高维稠密向量查询语言SQL向量相似度检索k-NN索引结构B 树HNSW、IVF、PQ、DiskANN适用场景精确匹配、事务语义搜索、推荐、RAG、多媒体检索常用向量数据库向量数据库数据存储方式索引类型优势特点使用场景部署形式Redis Stack内存磁盘HNSW、Flat高速读写、低延迟、支持 KV 和向量混合存储、内置向量检索实时推荐、在线问答、搜索增强单机/集群/DockerMilvus磁盘内存IVF、HNSW、ANNOY专业向量数据库支持大规模向量检索、高并发、分布式多模态检索、AI 搜索、图片/视频相似度单机/分布式Weaviate文档 向量HNSW开箱即用的语义搜索支持 GraphQL、向量 文本属性混合查询企业知识库、语义搜索、聊天机器人单机/CloudPinecone云托管HNSW、IVFSaaS 服务自动扩展管理简单无运维压力AI 应用向量搜索、推荐系统CloudQdrant内存 磁盘HNSW高性能、支持 REST/gRPC容易与 Python 集成文本/图像向量检索、AI 应用单机/集群/DockerFAISS内存为主IVF、PQ、HNSW高性能向量索引库GPU 加速适合大规模离线检索大规模向量计算、科研实验、批量搜索Python/C 库向量检索核心概念给定一个「查询向量」在庞大的向量集合里毫秒级找出与其最相似的 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( hostlocalhost, port6379, passwordNone, decode_responsesFalse # 存向量要关掉 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} ) ], definitionIndexDefinition(prefix[doc:]) # 建议加上前缀 ) print(✅ 已创建向量索引) # 写入一条数据 def insert_text(text: str): 调用通义千问 embedding 接口并将文本及其向量表示写入 Redis。 参数: text (str): 需要转换为向量并存储的原始文本内容。 # 调用多模态 embedding 接口获取文本向量 resp dashscope.MultiModalEmbedding.call( modelmultimodal-embedding-v1, input[{text: text}] ) if resp.status_code HTTPStatus.OK: # 提取 embedding 向量并转换为字节格式 embedding resp.output[embeddings][0][embedding] vector np.array(embedding, dtypenp.float32).tobytes() # 构造 Redis 键名 key fdoc:{resp.request_id} # 将文本和向量写入 Redis Hash 结构中 redis_client.hset(key, mapping{ text: text, embedding: vector }) print(f✅ 已写入 Rediskey{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( modelmultimodal-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], dtypenp.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(我喜欢用小米)执行结果✅ 已创建向量索引 ✅ 已写入 Rediskeydoc:e86bd66c-295a-92fe-9174-c799a38da841, 向量维度1024 ✅ 已写入 Rediskeydoc:6c3b9196-7d85-9d6e-95f4-a0eddde938bd, 向量维度1024 ✅ 已写入 Rediskeydoc:73ce1dad-cedb-95b8-bd5e-b5b2c4cd3f63, 向量维度1024 与 我喜欢用小米 最相似的 1 条 1. 我喜欢用苹果手机查看数据库-c799a38da841, 向量维度1024✅ 已写入 Rediskeydoc:6c3b9196-7d85-9d6e-95f4-a0eddde938bd, 向量维度1024✅ 已写入 Rediskeydoc:73ce1dad-cedb-95b8-bd5e-b5b2c4cd3f63, 向量维度1024 与 ‘我喜欢用小米’ 最相似的 1 条我喜欢用苹果手机查看数据库 
