本篇我们继续 RAG 实战先讲解稠密向量与稀疏向量的核心概念再一步步实操 Milvus 向量库完成数据表创建、文档入库添加数据、向量删除最后实现单路检索与稠密‑稀疏混合检索搭配 RRF 重排对接大模型完成问答。一、向量存储核心概念稠密向量 VS 稀疏向量向量是 RAG 知识库最核心的数据向量数据库 Milvus 支持两种向量类型二者各司其职搭配使用就是现在主流的混合检索方案。1. 稠密向量 (Dense Vector)特征向量内绝大部分数值不为 0向量长度固定由嵌入模型决定BGE‑M3 输出维度为 1024核心能力捕捉整体语义擅长理解文字背后的含义不局限于字面关键词。业务案例用户提问电动车跑不远文档内容写电瓶车续航差。两个句子用词完全不一样但是语义相同稠密向量就可以识别出二者相似度。配套索引HNSW分层导航小世界索引HNSW 查询逻辑从最高层入口出发贪心找最近邻居一层一层往下钻最后在 Layer0 精细搜索举例假如从武汉开高速到西安小区️ Layer2 国家级高速层顶层 起点是高速入口当前层只有几个大城市节点。 贪心规则永远往离目的地更近的节点开。 从武汉高速入口直接一跳开到【西安枢纽】。 Layer1 省级国道层 现在已经落到西安。 在这一层在西安周边的城市 / 区县之间继续找最近的点开到雁塔区。 国道层也看不到你家小区找不到更近的点继续下钻到 Layer0。️ Layer0 城市街道底层存放全部地点 到了城市街道这里有全国所有地点包含你家小区。 在这一层慢慢遍历周边街道小区找到你的家返回结果。✅总结路线 武汉 →(高速Layer2)→西安枢纽 →(国道Layer1)→雁塔区 →(街道Layer0)→你家小区如果暴力搜索 FLAT相当于不使用高速全国所有街道一家一家挨个找遍历全国每一个地点慢到爆炸。 HNSW 靠高层高速做远距离跳跃避免挨个遍历全部数据。优缺点✅ 优势语义理解能力强、模糊匹配效果好❌ 缺点向量维度高对比运算耗时更长2. 稀疏向量 (Sparse Vector)特征向量绝大多数位置数值为 0只有少量关键词带有权重值向量长度不固定。核心能力捕捉关键词、词频特征只认具体文字不理解深层语义。业务案例搜索苹果手机可以精准命中包含苹果、手机关键词的文档过滤掉介绍水果苹果的无关内容。配套索引SPARSE_INVERTED_INDEX倒排索引大白话理解原先是从文档里找词现在是从词中找文档例如找标和量从这两个字找相关文档然后取有这个文字的文档的交集例如用户查询句子标量拆解出两个词标、量如果不用倒排会怎么样 正向索引做法遍历全部 doc1、doc2、doc3逐个看文档里面有没有 “标” 和 “量”。文档数量巨大时遍历全部文档速度极慢。 倒排直接通过词快速定位候选文档不用扫全部数据。拿着词标去倒排索引查得到文档集合{doc1doc2}拿着词量去倒排索引查得到文档集合{doc1doc3}求两个集合交集同时包含两个词的文档 → doc1再把 doc1 里面两个词的权重做累加算出稀疏向量相似度分数作为最终得分返回。优缺点✅ 优势存储空间占用小、关键词召回速度极快❌ 缺点无法识别同义词、近义词语义模糊场景召回差3. 实战最佳方案稠密 稀疏「混合检索」向量类型擅长任务检索索引口诀总结稠密向量语义相似度匹配HNSW懂意思不认词稀疏向量关键词精准召回倒排索引认词语不懂意RAG 开发思路提炼 稠密向量负责找「意思相近」的文档稀疏向量负责找「关键词命中」的文档两份召回结果经过重排融合兼顾语义召回和关键词精准度大幅度减少漏召回二、代码实战 1BGE‑M3模型 生成稠密 稀疏向量BGE‑M3 是当下热门的多模态嵌入模型可以一次性同时输出稠密向量 稀疏向量无需分别调用两个模型。 查看稠密向量和 稀疏向量 -基于BGE-m3模型 from FlagEmbedding import BGEM3FlagModel #1.创建模型对象 model BGEM3FlagModel(./assets/models/bge-m3) #2.对数据进行编码 result model.encode( [标量字段通常用来存储一些元数据并可以在搜索时通过元数据进行过滤], return_denseTrue,# 是否返回稠密向量 return_sparseTrue, # 是否返回稀疏向量 ) print(result) #3.稀疏向量的长度不固定, 稠密向量的长度固定的 [和模型的维度有关]运行输出解读{ dense_vecs: array([[-0.02839016, -0.05221425,...]]), #稠密向量固定长度1024 lexical_weights: [defaultdict(...,{6: 0.0647...})], #稀疏向量关键词编号权重长度不固定 colbert_vecs: None }✨知识点总结稠密向量dense_vecs数组长度 模型维度1024固定不变稀疏向量lexical_weights字典仅保存有意义关键词及其权重长度不固定三、代码实战 2Milvus 创建向量数据表创建 Milvus 表完整流程连接 Milvus 实例 → 设计数据表 Schema 字段 → 创建向量索引 → 新建集合 (数据表)数据表字段规划字段名字段类型作用idINT64 (主键)文档块唯一编号开启自动生成 IDtextVARCHAR存储切分后的原始文本内容metadataJSON存储文档来源路径、文本块起始下标等附加信息vectorFLOAT_VECTOR (1024 维)稠密向量字段sparse_vectorSPARSE_FLOAT_VECTOR稀疏向量字段完整建表代码 创建向量数据库中的表 1.创建milvus实例-连接数据库 2.设置表的约束[字段] 3.设置表中索引[稠密向量和稀疏向量服务] 4.创建表 from pymilvus import MilvusClient, DataType #1 2 5 步必会 #创建milvus实例-连接数据库 def get_milvus_client(): #1.创建连接数据库的实例 client MilvusClient( uri http://127.0.0.1:19530, ) #2.查看数据库中有哪些表 collections client.list_collections() print(collections) return client #测试函数 #get_milvus_client() #3.设置表的约束[字段] def build_schem(): return MilvusClient().create_schema( auto_id True,#自动为id字段赋值 ).add_field( #id字段类型为整数是主键 field_name id, datatypeDataType.INT64,is_primaryTrue ).add_field( # 添加 text字段 类型为字符串,最大长度为1500 field_nametext,datatypeDataType.VARCHAR,max_length1500 ).add_field( #添加metadata字段类型为json field_name metadata,datatypeDataType.JSON ).add_field( #添加稠密向量字段类型为浮点数向量维度1024 field_namevector,datatypeDataType.FLOAT_VECTOR,dim 1024, ).add_field( #添加稀疏向量字段类型为稀疏向量类型 field_namesparse_vector,datatypeDataType.SPARSE_FLOAT_VECTOR ) #build_schem() #4.设置表中索引[稠密向量和稀疏向量服务] def build_index(): from pymilvus import MilvusClient, DataType index_params MilvusClient.prepare_index_params() index_params.add_index( #4.1稠密向量索引 field_namevector, # 建立索引的字段 index_typeHNSW, # 索引类型 metric_typeL2, # 向量相似度度量方式 ) index_params.add_index( #4.2稀疏向量索引 field_namesparse_vector, index_typeSPARSE_INVERTED_INDEX, metric_typeIP, ) return index_params #build_index() #5创建表的函数,函数里面一个client传入get_milvus_client()第二个参数是默认值不用传 def create_collection(client,collection_name demo_collection): #5.1调用创建表的函数设置表名和约束索引 client.create_collection( collection_name collection_name, schema build_schem(), index_params build_index() ) print(client.list_collections()) #6.调用函数 create_collection(get_milvus_client())运行结果如图所示四、代码实战 3Milvus添加向量数据完整流程加载本地 Word 文档 → 递归文本切分 → BGE‑M3 生成双向量 → 组装数据 → 批量插入向量库#往表中插入数据 from langchain_community.document_loaders import UnstructuredWordDocumentLoader from langchain_text_splitters import RecursiveCharacterTextSplitter from pymilvus import MilvusClient from FlagEmbedding import BGEM3FlagModel #获取连接数据库的的对象 def get_milvus_client(): #创建连接数据库的实例 client MilvusClient( urihttp://127.0.0.1:19530, ) return client #插入数据的函数(括号里面是连接对象和数据库名称)冒号约束形参类型告知这个形参是什么类型 def insert_data(client:MilvusClient,collection_name): #1.读取word文档里面需要填文件路径和模式 document UnstructuredWordDocumentLoader( file_path ../a_retrieval/assets/sample.docx #模式使用默认模式-single ).load() #2.对文档进行切分(递归切分) text_Splitter RecursiveCharacterTextSplitter( separators[\n\n, \n, 。], # 分隔符列表 chunk_size400, # 每个块的最大长度 chunk_overlap50, # 每个块重叠的长度 length_functionlen, # 可选计算文本长度的函数默认为字符串长度可自定义函数来实现按 token 数切分 add_start_indexTrue, # 可选块的元数据中添加此块起始索引 ) #得到切分后的文档列表 document_list text_Splitter.split_documents(document) #因为内容太多了把前20个chunk块存到document_list 中 document_list document_list[0:20] #print(document_list[0:20]) #insert_data(get_milvus_client(),demo_collection) 结果如下 [Document(metadata{source: ../a_retrieval/assets/sample.docx, start_index: 0}, page_content中华人民共和国民法典 文件里面有两个属性metadata存文件的位置和起始索引 page_content-Document对象.page_content metadata中的source-Document对象的metadate[source] #3.使用嵌入模型切分的文档内容转为向量 #创建嵌入模型对象 modelBGEM3FlagModel(../a_retrieval/assets/models/bge-m3) #对数据进行编码 all_vectors model.encode( #这里可以用列表推导式 #[对对象的操作 for 来源中的对象 in 来源 判断] [doc.page_content for doc in document_list], return_dense True , #是否返回稠密向量 return_sparse True, #是否返回稀疏向量 ) #4.对向量做分类 #4.1稠密向量 dense_vectors all_vectors[dense_vecs]#dense_vecs的类型是array对象 #4.2存稀疏向量 sparse_vectors all_vectors[lexical_weights]#lexical_weights的类型是defaultdict #5.对数据做处理-准备一个空列表把切分文本和其对应的稠密向量和稀疏向量作为一个整体元素存到刚才的空列表中 data_list [] for doc,dense,sparse in zip(document_list,dense_vectors,sparse_vectors): #把doc文本,dense稠密向量,sprase稀疏向量 存入到datalist data_list.append( { text: doc.page_content , # 切分的文档内容 metadata: doc.metadata, # 文件地址和切分的起始索引 vector: dense , #稠密向量 sparse_vector: sparse ,#稀疏向量 } ) #6.添加数据 client.insert( collection_namecollection_name,#关键字调用传参 datadata_list, #这里的data_list是data_list [] ) # insert_data(get_milvus_client(),demo_collection)结果分析每一条文档切片都会同时保存三份配套的数据metadata元数据记录这条文本来自哪个源文件路径用于后续溯源文档来源。vector稠密向量也就是稠密向量数组负责语义检索用来匹配意思相近的内容。sparse_vector稀疏向量稀疏向量以字典格式存储负责关键词检索精准命中文字。重点结论每一行一条文档切片一一对应一组稠密向量 一组稀疏向量。双向量成对绑定存入数据库后面就可以开启混合检索同时兼顾语义匹配、关键词命中。对应代码# 循环一一绑定文档切片、稠密向量、稀疏向量、元数据 data_list [] for doc,dense,sparse in zip(document_list,dense_vectors,sparse_vectors): data_list.append( { text: doc.page_content , # 切分的文档原文 metadata: doc.metadata, # 文件路径、文本起始下标 vector: dense , # 稠密向量语义检索 sparse_vector: sparse # 稀疏向量关键词检索 } ) # 将组装好的数据批量插入Milvus client.insert( collection_namecollection_name, datadata_list )循环里面的每一条字典数据就对应可视化面板里一行入库记录metadata→ 控制台metadata列vector→ 控制台vector稠密向量列sparse_vector→控制台sparse_vector稀疏向量列每一块切分出来的文档都会生成一对一绑定的稠密向量与稀疏向量同时存入向量数据库。Milvus 数据添加执行流程首先调用get_milvus_client()获取 Milvus 数据库的连接客户端再传入集合名称demo_collection两个参数一起传给insert_data入库函数进入函数内部后collection_name参数继续向下传递接着组装每一条文档数据稠密向量、稀疏向量、元数据、原文文本打包成data_list在client.insert()方法中把collection_name和data_list传入最终执行将向量文档数据写入 Milvus 指定的集合中。五、Milvus 删除向量数据#从表中删除数据 from pymilvus import MilvusClient #1.获取连接数据库的的对象 def get_milvus_client(): #创建连接数据库的实例 client MilvusClient( urihttp://127.0.0.1:19530, ) return client #2.删除数据的函数 def delete_data(client: MilvusClient,collection_name): #客户端对象去调用delete函数 client.delete( collection_namecollection_name,#表名 #过滤条件 filterid in [468672494323370209]#过滤条件:根据id来删 ) delete_data(get_milvus_client(),demo_collection)此记录468672494323370209已删除六、代码实战 4Milvus 检索单路检索 稠密稀疏混合检索RAG 检索完整链路用户输入问题 → BGE‑M3 生成查询向量 → Milvus 向量库召回文档 → RRFRanker 重排结果 → 提取文档上下文交给大模型生成答案 测试Milvus的检索 -查询 from langchain_openai import ChatOpenAI from pymilvus import MilvusClient, AnnSearchRequest, RRFRanker from FlagEmbedding import BGEM3FlagModel #需要把查询的内容转成向量再从向量数据库进行匹配 #1.获取数据库对象 #1.获取连接数据库的的对象 def get_milvus_client(): #创建连接数据库的实例 client MilvusClient( urihttp://127.0.0.1:19530, ) return client #2.把用户的查询内容转成 向量-为了去向量数据库中做查询匹配 def query_encode(query):#用户的查询内容 #创建模型对象 model BGEM3FlagModel(../a_retrieval/assets/models/bge-m3) #对数据进行编码 #all_vectores是一个字典根据键找值 all_vectores model.encode( query , return_denseTrue, #是否返回稠密向量 return_sparseTrue, #是否返回稀疏向量 ) #python返回多个数据数据会被打包成一个元组存稠密向量和稀疏向量的列表对象 return all_vectores[dense_vecs],all_vectores[lexical_weights] #3.查询稠密向量,里面传入问题数据库对象数据库的名字 def search_dense_vector(query:str,client:MilvusClient,collection_name:str): #从编码函数query_encode中取出返回值元组存储的稠密向量 #dense_vecs,_意思是一个变量dense_vecs一个变量是_ dense_vecs,_ query_encode(query) #查询 result client.search( collection_namecollection_name, #表名 data[dense_vecs],#用户输入的查询数据转换成的稠密向量 anns_fieldvector,#拿dense_vecs去查询数据库表中的哪一个字段-vector要和表中稠密向量的字段相同 limit5,#代表top k search_params{metric_type:L2} ,#设置检索方式 output_fields[id,text,metadata] #输出内容 ) print(result) #search_dense_vector(监护人的职责是什么,get_milvus_client(),demo_collection) #4.根据稀疏向量查询表中 匹配的数据 def search_sparse_vector(query:str,client:MilvusClient,collection_name:str): #从编码函数query_encode中取出返回值元组存储的稀疏向量 _,sparse_vecs query_encode(query) # 查询 result client.search( collection_namecollection_name, # 表名 data[sparse_vecs], # 用户输入的查询数据转换成的 稀疏向量 anns_fieldsparse_vector, # 拿sparse_vecs去查询数据库表中的哪一个字段 - sparse_vector 稀疏向量 limit5, # top k search_params{metric_type: IP}, # 设置检索方式 output_fields[id, text, metadata] ) print(result) #search_sparse_vector(监护人的职责是什么,get_milvus_client(),demo_collection) #两个方式总结 #稠密向量更懂语义看不懂词所以找语义相近的词进行返回 #稀疏向量更懂词不懂语义根据词来找结果 #5.混合向量检索 def hybrid_search(query : str,client : MilvusClient,collection_name : str): # 获取用户查询数据的 稠密向量和稀疏向量 集 dense_vecs , sparse_vecs query_encode(query) # 2次请求 - 一次是稠密向量 一次是稀疏向量 dense_req AnnSearchRequest( data [dense_vecs], anns_fieldvector, param {metric_type: L2}, limit5 ) sparse_req AnnSearchRequest( data[sparse_vecs], anns_fieldsparse_vector, param{metric_type: IP}, limit5 ) # 启动向量数据库的联合查询 results client.hybrid_search( collection_name collection_name, # 数据库表名 reqs [dense_req, sparse_req], # 要查询内容的 稠密向量和稀疏向量 limit 5, # top k ranker RRFRanker(k60), # 重排 output_fields [id,text,metadata] # 需要查询的字段 ) return results #hybrid_search(监护人的职责是什么,get_milvus_client(),demo_collection) #6.拿向量数据库中查询的结果 作为上下文和用户提示词提交给LLM-LLM生成最终的反馈内容 def main(query): #1.创建大语言模型对象 llm ChatOpenAI( model_namegpt-4o-mini ) #2.获取联合查询的结果集 results hybrid_search(query, get_milvus_client(), demo_collection) #3.把结果中有用的部分 整理成上下文 #join是连接意思从results里面取data,data是里面元素\n是换行把join里的连接起来 #data[entity]这个是往列表存的数据 content \n.join([data[entity][text] for data in results[0]]) #4.#提示词工程 message_list [ {role:system , content : 你是一个专业的法律问答机器人请根据上下文回答问题当上下文无法回答问题时请回答“根据上下文无法回答该问题}, {role:user , content : f根据以下上下文回答问题: {content}, 问题 : {query}} ] # 5.把提示词提交给大模型 RAG_Answers llm.invoke(message_list) print(RAG_Answers) main(监护人应该履行什么职责?) #它会把自己变成向量-去向量数据库做稠密和稀疏向量查询-从向量数据库拿到相应的结果-再把问题和提示词提交给大语言模型-大模型整理结果返回重点代码解释content ”\n”.join([data[”entity”][”text”] for data in results[0]])这行代码的作用就是从 Milvus 返回的检索结果里提取文档片段拼接成给大模型使用的参考上下文。我们逐层拆解results[0]取出混合检索命中的第一批最相关文档块for data in results[0]循环遍历每一条检索记录data[entity][text]文本内容嵌套存放在entity对象内部两层取值拿到原始文档片段。⚠️避坑提醒千万不要直接写data[text]外层字典没有 text 字段直接取值代码就会报错。\n.join( )将多条命中的文档片段用换行符隔开合并成一整段文本最后赋值给content变量这份拼接完成的上下文就可以连同用户问题一起送入大模型提示词。通俗来讲把向量库检索回来的资料挑出正文内容拼成一份完整参考资料交给大模型。图解RAG整体流程总结到这里我们就完整走完了RAG 检索增强生成的全链路实战。学AI大模型的正确顺序千万不要搞错了2026年AI风口已来各行各业的AI渗透肉眼可见超多公司要么转型做AI相关产品要么高薪挖AI技术人才机遇直接摆在眼前有往AI方向发展或者本身有后端编程基础的朋友直接冲AI大模型应用开发转岗超合适就算暂时不打算转岗了解大模型、RAG、Prompt、Agent这些热门概念能上手做简单项目也绝对是求职加分王给大家整理了超全最新的AI大模型应用开发学习清单和资料手把手帮你快速入门学习路线:✅大模型基础认知—大模型核心原理、发展历程、主流模型GPT、文心一言等特点解析✅核心技术模块—RAG检索增强生成、Prompt工程实战、Agent智能体开发逻辑✅开发基础能力—Python进阶、API接口调用、大模型开发框架LangChain等实操✅应用场景开发—智能问答系统、企业知识库、AIGC内容生成工具、行业定制化大模型应用✅项目落地流程—需求拆解、技术选型、模型调优、测试上线、运维迭代✅面试求职冲刺—岗位JD解析、简历AI项目包装、高频面试题汇总、模拟面经以上6大模块看似清晰好上手实则每个部分都有扎实的核心内容需要吃透我把大模型的学习全流程已经整理好了抓住AI时代风口轻松解锁职业新可能希望大家都能把握机遇实现薪资/职业跃迁这份完整版的大模型 AI 学习资料已经上传CSDN朋友们如果需要可以微信扫描下方CSDN官方认证二维码免费领取【保证100%免费】