本文是 refine-rag 系列教程的第八篇我们来学习向量存储的核心技术和 Milvus 数据库的使用。本文所有代码都在https://github.com/zonezoen/refine-rag往期系列文章• 007RAG 入门-向量嵌入与检索• 006RAG 入门-面试官问你RAG 为什么要切块• 005RAG 入门-LangChain读取表格数据目录• 前言• 为什么传统数据库搞不定向量搜索• 向量数据库横向对比• 为什么选择 Milvus• 快速搭建 Milvus• Milvus 核心概念• 实战示例• 学习路径前言前面我们学习了向量嵌入与检索知道如何把文本转成向量、如何计算相似度。但有个问题这些向量存在哪里如果只有几千条数据存在内存里就够了。但如果有几百万、几千万条数据呢这时候就需要专业的向量数据库。向量数据库就像是为向量量身定制的仓库管理系统它不仅能存储海量向量还能快速检索、高效管理。为什么传统数据库搞不定向量搜索上一篇文章提到向量本质上是一个浮点数数组是机器理解文本语义的方式。你可能会想既然向量就是数组那我存进 MySQL 的JSON字段或者直接存在内存List里不行吗答案是小规模可以大规模必死。当你的知识库达到万级、百万级甚至亿级时你会面临1.检索效率塌方计算量过大传统数据库查找数据是“相等判断”而向量搜索需要计算空间距离。如果你有 100 万条数据每搜一次都要全表扫一遍并计算相似度O(n) 复杂查询一次可能要几秒甚至几分钟。向量数据库通过ANN近似最近邻算法如 HNSW将搜索效率提升到了毫秒级O(log n) 复杂度。2.维度灾难内存压力一个 1536 维的向量OpenAI 标准看起来不大但几百万条叠在一起会瞬间撑爆普通的服务器内存。向量数据库专门优化了数据的压缩与加载策略。3.工程化缺失向量数据库不仅仅是“存数据”它还提供了成熟的 CRUD、多租户隔离、数据备份和水平扩展Scale-out能力这些是简单的内存库如 FAISS难以胜任的。核心对比传统数据库 vs 向量数据库传统数据库查找的是精准的数据而向量数据库则更适合模糊搜索。•传统数据库 (MySQL)• 你问“有没有编号为1024的商品”他秒回。• 你问“有没有长得像苹果的手机”他直接罢工”。•向量数据库 (Milvus)• 他会把所有数据转化为空间坐标告诉你“虽然我没找到完全一样的但有几个神似的数据。”代码层面的直观感受传统数据库MySQL侧重“准”-- 精确查询差一个字都搜不到 SELECT * FROM products WHERE category 手机 AND brand 苹果; -- 范围查询基于确定的数值边界 SELECT * FROM products WHERE price BETWEEN 5000 AND 8000;•特点擅长精确匹配基于 B-Tree 索引查询条件极其明确非黑即白。向量数据库Milvus侧重“像”# 相似度查询基于“语义距离” results client.search( collection_nameproducts, data[query_vector], # 搜索“长得像苹果的手机”对应的向量 limit10 )•特点擅长模糊搜索基于向量索引HNSW、IVF返回的是 Top-K 个“最相似”的候选者。向量数据库横向对比目前主流的向量数据库有很多我们来对比一下主流向量数据库对比数据库类型开源部署方式性能生态推荐度Milvus专业向量库是本地/云5/55/55/5Qdrant专业向量库是本地/云4/54/54/5Weaviate专业向量库是本地/云4/54/54/5Pinecone云服务否仅云5/54/53/5Chroma轻量级是本地3/53/53/5FAISS库非数据库是本地5/53/53/5pgvectorPostgreSQL 插件是本地/云3/54/53/5详细对比1. Milvus优点• 性能强大支持亿级向量• 索引算法丰富HNSW、IVF、DiskANN• 支持混合检索向量 标量过滤• 云原生架构易扩展• 社区活跃文档完善• 国内团队开发中文支持好缺点• 部署稍复杂需要 etcd、MinIO• 资源占用较高适用场景• 生产环境• 大规模数据建议百万级以上使用• 需要高性能和稳定性2. Qdrant优点• Rust 编写性能优秀• 部署简单单二进制文件• API 设计优雅• 支持 payload 过滤缺点• 社区相对较小• 中文文档较少• 大规模数据性能不如 Milvus适用场景• 中小规模项目• 快速原型开发• 喜欢 Rust 生态3. Weaviate优点• 内置 GraphQL API• 支持多模态搜索• 自带向量化模块• 语义搜索能力强缺点• 学习曲线陡峭• 资源占用高• 配置复杂适用场景• 知识图谱应用• 多模态搜索• 需要 GraphQL4. Pinecone优点• 完全托管无需运维• 性能优秀• 易用性好缺点• 闭源不可自部署• 费用较高• 数据在国外延迟问题适用场景• 不想自己运维• 预算充足• 海外用户5. Chroma优点• 极简设计易上手• 轻量级适合开发• 与 LangChain 集成好缺点• 性能一般• 不适合大规模生产• 功能相对简单适用场景• 学习和原型开发• 小规模应用10 万条• 快速验证想法6. FAISS优点• Facebook 出品算法先进• 性能极强• 支持 GPU 加速缺点• 不是数据库只是库• 没有持久化需自己实现• 没有分布式支持适用场景• 研究和实验• 需要 GPU 加速• 自己实现存储层7. pgvector优点• PostgreSQL 插件易集成• 利用现有 PG 生态• 事务支持缺点• 性能不如专业向量库• 索引算法有限• 大规模数据吃力适用场景• 已有 PostgreSQL• 数据量不大100 万• 需要事务支持为什么选择 Milvus综合考虑选择 Milvus 作为本教程的向量数据库原因如下1. 性能强大# Milvus 可以轻松处理亿级向量 # 查询延迟毫秒级 # QPS数千到数万2. 索引算法丰富索引类型适用场景性能FLAT小数据集完美准确率2/5IVF_FLAT中等数据集平衡4/5IVF_PQ大数据集内存受限4/5HNSW高性能查询5/5DiskANN超大规模磁盘存储4/53. 功能完善• 混合检索向量 标量过滤• 分区管理Partition• 多租户支持Database• 动态字段Dynamic Field• 全文检索BM25• 范围搜索Range Search• 分组搜索Group Search4. 生态完善# 与主流框架无缝集成 from langchain_milvus import Milvus from llama_index.vector_stores import MilvusVectorStore5. 国内支持好• 中文文档完善• 社区活跃国内团队• 国内访问速度快• 技术支持响应快6. 云原生架构Milvus 架构 ┌─────────────────────────────────────┐ │ Milvus Coordinator │ 协调层 ├─────────────────────────────────────┤ │ Query Node │ Data Node │ Index │ 计算层 ├─────────────────────────────────────┤ │ etcd │ MinIO │ Pulsar │ 存储层 └─────────────────────────────────────┘• 存储计算分离• 易于水平扩展• 高可用性7. 面试现在很多企业也是用的这套方案我们学习这些不就是奔着找工作去的吗直接朝着目标前进不好吗快速搭建 MilvusDocker Compose推荐文件docker-compose.ymlversion: 3.8 services: # ---------------------------------------------------------------- # 1. etcd - 元数据存储 # ---------------------------------------------------------------- etcd: image: quay.io/coreos/etcd:v3.5.5 container_name: milvus-etcd environment: - ETCD_AUTO_COMPACTION_MODErevision - ETCD_AUTO_COMPACTION_RETENTION1000 - ETCD_QUOTA_BACKEND_BYTES4294967296 - ETCD_SNAPSHOT_COUNT50000 volumes: - ./milvus_data/etcd:/etcd command: etcd -advertise-client-urlshttp://127.0.0.1:2379 -listen-client-urls http://0.0.0.0:2379 --data-dir /etcd healthcheck: test: [CMD, etcdctl, endpoint, health] interval: 30s timeout: 20s retries: 3 # ---------------------------------------------------------------- # 2. MinIO - 对象存储 # ---------------------------------------------------------------- minio: image: minio/minio:RELEASE.2023-03-20T20-16-18Z container_name: milvus-minio environment: MINIO_ACCESS_KEY: minioadmin MINIO_SECRET_KEY: minioadmin ports: - 9001:9001 - 9000:9000 volumes: - ./milvus_data/minio:/minio_data command: minio server /minio_data --console-address :9001 healthcheck: test: [CMD, curl, -f, http://localhost:9000/minio/health/live] interval: 30s timeout: 20s retries: 3 # ---------------------------------------------------------------- # 3. Milvus Standalone - 向量数据库 # ---------------------------------------------------------------- milvus-standalone: image: milvusdb/milvus:v2.5.4 container_name: milvus-standalone command: [milvus, run, standalone] security_opt: - seccomp:unconfined environment: ETCD_ENDPOINTS: etcd:2379 MINIO_ADDRESS: minio:9000 # ---------------------------------------------------------------- # 开启认证设置密码 # ---------------------------------------------------------------- # 取消下面两行注释即可开启用户认证 # COMMON_SECURITY_AUTHORIZATIONENABLED: true # 开启认证 # COMMON_SECURITY_DEFAULTROOTPASSWORD: your_password_here # 设置 root 用户密码 # # 开启后Python 连接时需要提供 token # client MilvusClient( # urihttp://localhost:19530, # tokenroot:your_password_here # ) volumes: - ./milvus_data/milvus:/var/lib/milvus healthcheck: test: [CMD, curl, -f, http://localhost:9091/healthz] interval: 30s start_period: 90s timeout: 20s retries: 3 ports: - 19530:19530 - 9091:9091 depends_on: - etcd - minio启动步骤# 1. 进入目录 cd rag/08-向量存储/ # 2. 启动所有服务 docker compose up -d # 3. 查看服务状态 docker compose ps # 4. 查看日志等待启动完成 docker compose logs -f milvus-standalone # 看到这行说明启动成功 # [INFO] Milvus Proxy successfully started验证安装from pymilvus import MilvusClient # 连接 Milvus client MilvusClient(urihttp://localhost:19530) # 列出数据库 databases client.list_databases() print(数据库列表:, databases) # [default] print(✓ Milvus 连接成功)停止服务# 停止服务保留数据 docker compose stop # 停止并删除容器保留数据 docker compose down # 停止并删除所有数据 docker compose down -v rm -rf milvus_data/服务端口说明服务端口说明Milvus19530gRPC 接口主要Milvus9091HTTP 监控接口MinIO9000对象存储 APIMinIO9001Web 控制台etcd2379元数据存储访问 MinIO 控制台浏览器打开http://localhost:9001• 用户名minioadmin• 密码minioadmin可以看到 Milvus 存储的向量数据。Milvus 核心概念1. Database数据库类比图书馆的不同楼层# 创建数据库 client.create_database(db_namemy_rag_system) # 切换数据库 client.use_database(db_namemy_rag_system) # 列出所有数据库 databases client.list_databases()用途• 隔离不同业务• 多租户管理2. Collection集合类比图书馆的书架# 创建集合快速模式 client.create_collection( collection_namedocuments, dimension128 # 向量维度 )特点• 必须包含向量字段• 类似 MySQL 的表3. Schema表结构类比书籍的目录格式from pymilvus import CollectionSchema, FieldSchema, DataType # 定义字段 fields [ FieldSchema(nameid, dtypeDataType.INT64, is_primaryTrue, auto_idTrue), FieldSchema(namevector, dtypeDataType.FLOAT_VECTOR, dim128), FieldSchema(nametext, dtypeDataType.VARCHAR, max_length1000), FieldSchema(namesource, dtypeDataType.VARCHAR, max_length100), ] # 创建 Schema schema CollectionSchema( fieldsfields, description文档集合, enable_dynamic_fieldTrue # 允许动态字段 ) # 创建集合 client.create_collection( collection_namedocuments, schemaschema )4. Partition分区类比书架的不同层# 创建分区 client.create_partition( collection_namedocuments, partition_nametech_docs ) # 插入数据到指定分区 client.insert( collection_namedocuments, data[...], partition_nametech_docs ) # 只在指定分区搜索速度更快 results client.search( collection_namedocuments, data[query_vector], partition_names[tech_docs] )优势• 提高搜索速度• 方便数据管理• 节省内存5. Index索引后续的文章会介绍如何选择索引类型和距离度量方式。类比图书馆的索引系统# 创建索引 index_params client.prepare_index_params() index_params.add_index( field_namevector, index_typeHNSW, # 索引类型 metric_typeL2, # 距离度量 params{ M: 16, efConstruction: 200 } ) client.create_index( collection_namedocuments, index_paramsindex_params )常用索引• FLAT小数据集暴力搜索• IVF_FLAT中等数据集• HNSW高性能• DiskANN超大规模6. Load/Release加载/释放类比从仓库搬到阅览室# 加载到内存必须 client.load_collection(collection_namedocuments) # 查询从内存读取快 results client.search(...) # 释放内存 client.release_collection(collection_namedocuments)为什么需要 load• 向量搜索需要大量计算• 从内存读取比磁盘快 100 倍• 兼顾持久化和性能实战示例完整示例黑神话悟空妖怪数据库文件a-working-sample.pyimport logging import pandas as pd from pymilvus import MilvusClient, DataType, FieldSchema, CollectionSchema from sentence_transformers import SentenceTransformer from tqdm import tqdm # 隐藏日志 logging.getLogger(pymilvus).setLevel(logging.CRITICAL) # —————————————— # 1. 准备数据 # —————————————— data_records [ { monster_id: BM001, monster_name: 虎先锋, location: 竹林关隘, difficulty: High, synonyms: 猛虎妖, 虎妖, description: 在竹林关卡中出现的猛虎型妖怪力量强大。 }, { monster_id: BM002, monster_name: 火猿, location: 火山洞窟, difficulty: Low, synonyms: 烈焰猿, 炎猿, description: 生活在火山洞窟的猿类妖怪只是插科打诨的小兵。 }, ] df pd.DataFrame(data_records) # —————————————— # 2. 连接 Milvus # —————————————— client MilvusClient(urihttp://localhost:19530) collection_name Wukong_Monsters # —————————————— # 3. 加载 Embedding 模型 # —————————————— print(加载 embedding 模型...) embedding_model SentenceTransformer(BAAI/bge-small-zh-v1.5) sample_embedding embedding_model.encode([示例文本])[0] vector_dim len(sample_embedding) print(f向量维度: {vector_dim}) # —————————————— # 4. 定义 Schema 并创建集合 # —————————————— fields [ FieldSchema(nameid, dtypeDataType.INT64, is_primaryTrue, auto_idTrue), FieldSchema(namevector, dtypeDataType.FLOAT_VECTOR, dimvector_dim), FieldSchema(namemonster_id, dtypeDataType.VARCHAR, max_length50), FieldSchema(namemonster_name, dtypeDataType.VARCHAR, max_length100), FieldSchema(namelocation, dtypeDataType.VARCHAR, max_length100), FieldSchema(namedifficulty, dtypeDataType.VARCHAR, max_length20), FieldSchema(namesynonyms, dtypeDataType.VARCHAR, max_length200), FieldSchema(namedescription, dtypeDataType.VARCHAR, max_length500), ] schema CollectionSchema( fields, descriptionWukong Monsters, enable_dynamic_fieldTrue ) if not client.has_collection(collection_name): client.create_collection(collection_namecollection_name, schemaschema) # —————————————— # 5. 创建索引 # —————————————— index_params client.prepare_index_params() index_params.add_index( field_namevector, index_typeAUTOINDEX, metric_typeL2 ) client.create_index(collection_namecollection_name, index_paramsindex_params) # —————————————— # 6. 插入数据 # —————————————— for start_idx in tqdm(range(len(df)), desc插入数据): row df.iloc[start_idx] # 准备向量文本 doc_parts [str(row[monster_name])] if row[synonyms]: doc_parts.append(f(别名{row[synonyms]})) if row[location]: doc_parts.append(f场景{row[location]}) if row[description]: doc_parts.append(f描述{row[description]}) doc_text .join(doc_parts) # 生成向量 embedding embedding_model.encode([doc_text])[0] # 插入数据 data_to_insert [{ vector: embedding.tolist(), monster_id: str(row[monster_id]), monster_name: str(row[monster_name]), location: str(row[location]), difficulty: str(row[difficulty]), synonyms: str(row[synonyms]), description: str(row[description]) }] client.insert(collection_namecollection_name, datadata_to_insert) # —————————————— # 7. 加载到内存 # —————————————— print(\n加载 Collection 到内存...) client.load_collection(collection_namecollection_name) print(✓ Collection 已加载) # —————————————— # 8. 向量搜索 # —————————————— search_query 高难度妖怪 search_embedding embedding_model.encode([search_query])[0] search_result client.search( collection_namecollection_name, data[search_embedding.tolist()], limit3, output_fields[monster_name, location, difficulty, synonyms] ) print(f\n搜索结果 {search_query}:) for hits in search_result: for hit in hits: print(f - {hit[entity]}) # —————————————— # 9. 条件查询 # —————————————— query_result client.query( collection_namecollection_name, filterdifficulty Low, output_fields[monster_name, location, difficulty, synonyms] ) print(f\n难度为 Low 的妖怪) for result in query_result: print(f - {result})运行结果加载 embedding 模型... 向量维度: 512 插入数据: 100%|████████████| 2/2 [00:0000:00, 10.5it/s] 加载 Collection 到内存... ✓ Collection 已加载 搜索结果 高难度妖怪: - {monster_name: 虎先锋, location: 竹林关隘, difficulty: High, synonyms: 猛虎妖, 虎妖} 难度为 Low 的妖怪 - {monster_name: 火猿, location: 火山洞窟, difficulty: Low, synonyms: 烈焰猿, 炎猿}学习路径1. 简易 RAG 学习2. LCEL 语法学习3. LangChain 读取数据1. LangChain 读取文本数据2. LangChain 读取图片数据3. LangChain 读取 PDF 数据4. LangChain 读取表格数据4. 文本切块5. 向量嵌入与检索6. 向量存储 ← 当前7. 检索前处理8. 索引优化9. 检索后处理10. 响应生成11. 系统评估项目地址本文所有代码示例都在 GitHub 开源https://github.com/zonezoen/refine-rag欢迎 Star 和 Fork一起学习 RAG 技术 008-01：RAG 入门-向量存储与企业级向量数据库 milvus
本文是 refine-rag 系列教程的第八篇我们来学习向量存储的核心技术和 Milvus 数据库的使用。本文所有代码都在https://github.com/zonezoen/refine-rag往期系列文章• 007RAG 入门-向量嵌入与检索• 006RAG 入门-面试官问你RAG 为什么要切块• 005RAG 入门-LangChain读取表格数据目录• 前言• 为什么传统数据库搞不定向量搜索• 向量数据库横向对比• 为什么选择 Milvus• 快速搭建 Milvus• Milvus 核心概念• 实战示例• 学习路径前言前面我们学习了向量嵌入与检索知道如何把文本转成向量、如何计算相似度。但有个问题这些向量存在哪里如果只有几千条数据存在内存里就够了。但如果有几百万、几千万条数据呢这时候就需要专业的向量数据库。向量数据库就像是为向量量身定制的仓库管理系统它不仅能存储海量向量还能快速检索、高效管理。为什么传统数据库搞不定向量搜索上一篇文章提到向量本质上是一个浮点数数组是机器理解文本语义的方式。你可能会想既然向量就是数组那我存进 MySQL 的JSON字段或者直接存在内存List里不行吗答案是小规模可以大规模必死。当你的知识库达到万级、百万级甚至亿级时你会面临1.检索效率塌方计算量过大传统数据库查找数据是“相等判断”而向量搜索需要计算空间距离。如果你有 100 万条数据每搜一次都要全表扫一遍并计算相似度O(n) 复杂查询一次可能要几秒甚至几分钟。向量数据库通过ANN近似最近邻算法如 HNSW将搜索效率提升到了毫秒级O(log n) 复杂度。2.维度灾难内存压力一个 1536 维的向量OpenAI 标准看起来不大但几百万条叠在一起会瞬间撑爆普通的服务器内存。向量数据库专门优化了数据的压缩与加载策略。3.工程化缺失向量数据库不仅仅是“存数据”它还提供了成熟的 CRUD、多租户隔离、数据备份和水平扩展Scale-out能力这些是简单的内存库如 FAISS难以胜任的。核心对比传统数据库 vs 向量数据库传统数据库查找的是精准的数据而向量数据库则更适合模糊搜索。•传统数据库 (MySQL)• 你问“有没有编号为1024的商品”他秒回。• 你问“有没有长得像苹果的手机”他直接罢工”。•向量数据库 (Milvus)• 他会把所有数据转化为空间坐标告诉你“虽然我没找到完全一样的但有几个神似的数据。”代码层面的直观感受传统数据库MySQL侧重“准”-- 精确查询差一个字都搜不到 SELECT * FROM products WHERE category 手机 AND brand 苹果; -- 范围查询基于确定的数值边界 SELECT * FROM products WHERE price BETWEEN 5000 AND 8000;•特点擅长精确匹配基于 B-Tree 索引查询条件极其明确非黑即白。向量数据库Milvus侧重“像”# 相似度查询基于“语义距离” results client.search( collection_nameproducts, data[query_vector], # 搜索“长得像苹果的手机”对应的向量 limit10 )•特点擅长模糊搜索基于向量索引HNSW、IVF返回的是 Top-K 个“最相似”的候选者。向量数据库横向对比目前主流的向量数据库有很多我们来对比一下主流向量数据库对比数据库类型开源部署方式性能生态推荐度Milvus专业向量库是本地/云5/55/55/5Qdrant专业向量库是本地/云4/54/54/5Weaviate专业向量库是本地/云4/54/54/5Pinecone云服务否仅云5/54/53/5Chroma轻量级是本地3/53/53/5FAISS库非数据库是本地5/53/53/5pgvectorPostgreSQL 插件是本地/云3/54/53/5详细对比1. Milvus优点• 性能强大支持亿级向量• 索引算法丰富HNSW、IVF、DiskANN• 支持混合检索向量 标量过滤• 云原生架构易扩展• 社区活跃文档完善• 国内团队开发中文支持好缺点• 部署稍复杂需要 etcd、MinIO• 资源占用较高适用场景• 生产环境• 大规模数据建议百万级以上使用• 需要高性能和稳定性2. Qdrant优点• Rust 编写性能优秀• 部署简单单二进制文件• API 设计优雅• 支持 payload 过滤缺点• 社区相对较小• 中文文档较少• 大规模数据性能不如 Milvus适用场景• 中小规模项目• 快速原型开发• 喜欢 Rust 生态3. Weaviate优点• 内置 GraphQL API• 支持多模态搜索• 自带向量化模块• 语义搜索能力强缺点• 学习曲线陡峭• 资源占用高• 配置复杂适用场景• 知识图谱应用• 多模态搜索• 需要 GraphQL4. Pinecone优点• 完全托管无需运维• 性能优秀• 易用性好缺点• 闭源不可自部署• 费用较高• 数据在国外延迟问题适用场景• 不想自己运维• 预算充足• 海外用户5. Chroma优点• 极简设计易上手• 轻量级适合开发• 与 LangChain 集成好缺点• 性能一般• 不适合大规模生产• 功能相对简单适用场景• 学习和原型开发• 小规模应用10 万条• 快速验证想法6. FAISS优点• Facebook 出品算法先进• 性能极强• 支持 GPU 加速缺点• 不是数据库只是库• 没有持久化需自己实现• 没有分布式支持适用场景• 研究和实验• 需要 GPU 加速• 自己实现存储层7. pgvector优点• PostgreSQL 插件易集成• 利用现有 PG 生态• 事务支持缺点• 性能不如专业向量库• 索引算法有限• 大规模数据吃力适用场景• 已有 PostgreSQL• 数据量不大100 万• 需要事务支持为什么选择 Milvus综合考虑选择 Milvus 作为本教程的向量数据库原因如下1. 性能强大# Milvus 可以轻松处理亿级向量 # 查询延迟毫秒级 # QPS数千到数万2. 索引算法丰富索引类型适用场景性能FLAT小数据集完美准确率2/5IVF_FLAT中等数据集平衡4/5IVF_PQ大数据集内存受限4/5HNSW高性能查询5/5DiskANN超大规模磁盘存储4/53. 功能完善• 混合检索向量 标量过滤• 分区管理Partition• 多租户支持Database• 动态字段Dynamic Field• 全文检索BM25• 范围搜索Range Search• 分组搜索Group Search4. 生态完善# 与主流框架无缝集成 from langchain_milvus import Milvus from llama_index.vector_stores import MilvusVectorStore5. 国内支持好• 中文文档完善• 社区活跃国内团队• 国内访问速度快• 技术支持响应快6. 云原生架构Milvus 架构 ┌─────────────────────────────────────┐ │ Milvus Coordinator │ 协调层 ├─────────────────────────────────────┤ │ Query Node │ Data Node │ Index │ 计算层 ├─────────────────────────────────────┤ │ etcd │ MinIO │ Pulsar │ 存储层 └─────────────────────────────────────┘• 存储计算分离• 易于水平扩展• 高可用性7. 面试现在很多企业也是用的这套方案我们学习这些不就是奔着找工作去的吗直接朝着目标前进不好吗快速搭建 MilvusDocker Compose推荐文件docker-compose.ymlversion: 3.8 services: # ---------------------------------------------------------------- # 1. etcd - 元数据存储 # ---------------------------------------------------------------- etcd: image: quay.io/coreos/etcd:v3.5.5 container_name: milvus-etcd environment: - ETCD_AUTO_COMPACTION_MODErevision - ETCD_AUTO_COMPACTION_RETENTION1000 - ETCD_QUOTA_BACKEND_BYTES4294967296 - ETCD_SNAPSHOT_COUNT50000 volumes: - ./milvus_data/etcd:/etcd command: etcd -advertise-client-urlshttp://127.0.0.1:2379 -listen-client-urls http://0.0.0.0:2379 --data-dir /etcd healthcheck: test: [CMD, etcdctl, endpoint, health] interval: 30s timeout: 20s retries: 3 # ---------------------------------------------------------------- # 2. MinIO - 对象存储 # ---------------------------------------------------------------- minio: image: minio/minio:RELEASE.2023-03-20T20-16-18Z container_name: milvus-minio environment: MINIO_ACCESS_KEY: minioadmin MINIO_SECRET_KEY: minioadmin ports: - 9001:9001 - 9000:9000 volumes: - ./milvus_data/minio:/minio_data command: minio server /minio_data --console-address :9001 healthcheck: test: [CMD, curl, -f, http://localhost:9000/minio/health/live] interval: 30s timeout: 20s retries: 3 # ---------------------------------------------------------------- # 3. Milvus Standalone - 向量数据库 # ---------------------------------------------------------------- milvus-standalone: image: milvusdb/milvus:v2.5.4 container_name: milvus-standalone command: [milvus, run, standalone] security_opt: - seccomp:unconfined environment: ETCD_ENDPOINTS: etcd:2379 MINIO_ADDRESS: minio:9000 # ---------------------------------------------------------------- # 开启认证设置密码 # ---------------------------------------------------------------- # 取消下面两行注释即可开启用户认证 # COMMON_SECURITY_AUTHORIZATIONENABLED: true # 开启认证 # COMMON_SECURITY_DEFAULTROOTPASSWORD: your_password_here # 设置 root 用户密码 # # 开启后Python 连接时需要提供 token # client MilvusClient( # urihttp://localhost:19530, # tokenroot:your_password_here # ) volumes: - ./milvus_data/milvus:/var/lib/milvus healthcheck: test: [CMD, curl, -f, http://localhost:9091/healthz] interval: 30s start_period: 90s timeout: 20s retries: 3 ports: - 19530:19530 - 9091:9091 depends_on: - etcd - minio启动步骤# 1. 进入目录 cd rag/08-向量存储/ # 2. 启动所有服务 docker compose up -d # 3. 查看服务状态 docker compose ps # 4. 查看日志等待启动完成 docker compose logs -f milvus-standalone # 看到这行说明启动成功 # [INFO] Milvus Proxy successfully started验证安装from pymilvus import MilvusClient # 连接 Milvus client MilvusClient(urihttp://localhost:19530) # 列出数据库 databases client.list_databases() print(数据库列表:, databases) # [default] print(✓ Milvus 连接成功)停止服务# 停止服务保留数据 docker compose stop # 停止并删除容器保留数据 docker compose down # 停止并删除所有数据 docker compose down -v rm -rf milvus_data/服务端口说明服务端口说明Milvus19530gRPC 接口主要Milvus9091HTTP 监控接口MinIO9000对象存储 APIMinIO9001Web 控制台etcd2379元数据存储访问 MinIO 控制台浏览器打开http://localhost:9001• 用户名minioadmin• 密码minioadmin可以看到 Milvus 存储的向量数据。Milvus 核心概念1. Database数据库类比图书馆的不同楼层# 创建数据库 client.create_database(db_namemy_rag_system) # 切换数据库 client.use_database(db_namemy_rag_system) # 列出所有数据库 databases client.list_databases()用途• 隔离不同业务• 多租户管理2. Collection集合类比图书馆的书架# 创建集合快速模式 client.create_collection( collection_namedocuments, dimension128 # 向量维度 )特点• 必须包含向量字段• 类似 MySQL 的表3. Schema表结构类比书籍的目录格式from pymilvus import CollectionSchema, FieldSchema, DataType # 定义字段 fields [ FieldSchema(nameid, dtypeDataType.INT64, is_primaryTrue, auto_idTrue), FieldSchema(namevector, dtypeDataType.FLOAT_VECTOR, dim128), FieldSchema(nametext, dtypeDataType.VARCHAR, max_length1000), FieldSchema(namesource, dtypeDataType.VARCHAR, max_length100), ] # 创建 Schema schema CollectionSchema( fieldsfields, description文档集合, enable_dynamic_fieldTrue # 允许动态字段 ) # 创建集合 client.create_collection( collection_namedocuments, schemaschema )4. Partition分区类比书架的不同层# 创建分区 client.create_partition( collection_namedocuments, partition_nametech_docs ) # 插入数据到指定分区 client.insert( collection_namedocuments, data[...], partition_nametech_docs ) # 只在指定分区搜索速度更快 results client.search( collection_namedocuments, data[query_vector], partition_names[tech_docs] )优势• 提高搜索速度• 方便数据管理• 节省内存5. Index索引后续的文章会介绍如何选择索引类型和距离度量方式。类比图书馆的索引系统# 创建索引 index_params client.prepare_index_params() index_params.add_index( field_namevector, index_typeHNSW, # 索引类型 metric_typeL2, # 距离度量 params{ M: 16, efConstruction: 200 } ) client.create_index( collection_namedocuments, index_paramsindex_params )常用索引• FLAT小数据集暴力搜索• IVF_FLAT中等数据集• HNSW高性能• DiskANN超大规模6. Load/Release加载/释放类比从仓库搬到阅览室# 加载到内存必须 client.load_collection(collection_namedocuments) # 查询从内存读取快 results client.search(...) # 释放内存 client.release_collection(collection_namedocuments)为什么需要 load• 向量搜索需要大量计算• 从内存读取比磁盘快 100 倍• 兼顾持久化和性能实战示例完整示例黑神话悟空妖怪数据库文件a-working-sample.pyimport logging import pandas as pd from pymilvus import MilvusClient, DataType, FieldSchema, CollectionSchema from sentence_transformers import SentenceTransformer from tqdm import tqdm # 隐藏日志 logging.getLogger(pymilvus).setLevel(logging.CRITICAL) # —————————————— # 1. 准备数据 # —————————————— data_records [ { monster_id: BM001, monster_name: 虎先锋, location: 竹林关隘, difficulty: High, synonyms: 猛虎妖, 虎妖, description: 在竹林关卡中出现的猛虎型妖怪力量强大。 }, { monster_id: BM002, monster_name: 火猿, location: 火山洞窟, difficulty: Low, synonyms: 烈焰猿, 炎猿, description: 生活在火山洞窟的猿类妖怪只是插科打诨的小兵。 }, ] df pd.DataFrame(data_records) # —————————————— # 2. 连接 Milvus # —————————————— client MilvusClient(urihttp://localhost:19530) collection_name Wukong_Monsters # —————————————— # 3. 加载 Embedding 模型 # —————————————— print(加载 embedding 模型...) embedding_model SentenceTransformer(BAAI/bge-small-zh-v1.5) sample_embedding embedding_model.encode([示例文本])[0] vector_dim len(sample_embedding) print(f向量维度: {vector_dim}) # —————————————— # 4. 定义 Schema 并创建集合 # —————————————— fields [ FieldSchema(nameid, dtypeDataType.INT64, is_primaryTrue, auto_idTrue), FieldSchema(namevector, dtypeDataType.FLOAT_VECTOR, dimvector_dim), FieldSchema(namemonster_id, dtypeDataType.VARCHAR, max_length50), FieldSchema(namemonster_name, dtypeDataType.VARCHAR, max_length100), FieldSchema(namelocation, dtypeDataType.VARCHAR, max_length100), FieldSchema(namedifficulty, dtypeDataType.VARCHAR, max_length20), FieldSchema(namesynonyms, dtypeDataType.VARCHAR, max_length200), FieldSchema(namedescription, dtypeDataType.VARCHAR, max_length500), ] schema CollectionSchema( fields, descriptionWukong Monsters, enable_dynamic_fieldTrue ) if not client.has_collection(collection_name): client.create_collection(collection_namecollection_name, schemaschema) # —————————————— # 5. 创建索引 # —————————————— index_params client.prepare_index_params() index_params.add_index( field_namevector, index_typeAUTOINDEX, metric_typeL2 ) client.create_index(collection_namecollection_name, index_paramsindex_params) # —————————————— # 6. 插入数据 # —————————————— for start_idx in tqdm(range(len(df)), desc插入数据): row df.iloc[start_idx] # 准备向量文本 doc_parts [str(row[monster_name])] if row[synonyms]: doc_parts.append(f(别名{row[synonyms]})) if row[location]: doc_parts.append(f场景{row[location]}) if row[description]: doc_parts.append(f描述{row[description]}) doc_text .join(doc_parts) # 生成向量 embedding embedding_model.encode([doc_text])[0] # 插入数据 data_to_insert [{ vector: embedding.tolist(), monster_id: str(row[monster_id]), monster_name: str(row[monster_name]), location: str(row[location]), difficulty: str(row[difficulty]), synonyms: str(row[synonyms]), description: str(row[description]) }] client.insert(collection_namecollection_name, datadata_to_insert) # —————————————— # 7. 加载到内存 # —————————————— print(\n加载 Collection 到内存...) client.load_collection(collection_namecollection_name) print(✓ Collection 已加载) # —————————————— # 8. 向量搜索 # —————————————— search_query 高难度妖怪 search_embedding embedding_model.encode([search_query])[0] search_result client.search( collection_namecollection_name, data[search_embedding.tolist()], limit3, output_fields[monster_name, location, difficulty, synonyms] ) print(f\n搜索结果 {search_query}:) for hits in search_result: for hit in hits: print(f - {hit[entity]}) # —————————————— # 9. 条件查询 # —————————————— query_result client.query( collection_namecollection_name, filterdifficulty Low, output_fields[monster_name, location, difficulty, synonyms] ) print(f\n难度为 Low 的妖怪) for result in query_result: print(f - {result})运行结果加载 embedding 模型... 向量维度: 512 插入数据: 100%|████████████| 2/2 [00:0000:00, 10.5it/s] 加载 Collection 到内存... ✓ Collection 已加载 搜索结果 高难度妖怪: - {monster_name: 虎先锋, location: 竹林关隘, difficulty: High, synonyms: 猛虎妖, 虎妖} 难度为 Low 的妖怪 - {monster_name: 火猿, location: 火山洞窟, difficulty: Low, synonyms: 烈焰猿, 炎猿}学习路径1. 简易 RAG 学习2. LCEL 语法学习3. LangChain 读取数据1. LangChain 读取文本数据2. LangChain 读取图片数据3. LangChain 读取 PDF 数据4. LangChain 读取表格数据4. 文本切块5. 向量嵌入与检索6. 向量存储 ← 当前7. 检索前处理8. 索引优化9. 检索后处理10. 响应生成11. 系统评估项目地址本文所有代码示例都在 GitHub 开源https://github.com/zonezoen/refine-rag欢迎 Star 和 Fork一起学习 RAG 技术