LangChain RAG 实战 | 稠密稀疏向量、Milvus 建库、增删检索数据

本篇我们继续 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，逐个看文档里面有没有 “标” 和 “量”。文档数量巨大时，遍历全部文档速度极慢。 倒排直接通过词快速定位候选文档，不用扫全部数据。

1. 拿着词标去倒排索引查：得到文档集合：{doc1，doc2}
2. 拿着词量去倒排索引查：得到文档集合：{doc1，doc3}
3. 求两个集合交集：同时包含两个词的文档 → doc1
4. 再把 doc1 里面两个词的权重做累加，算出稀疏向量相似度分数，作为最终得分返回。

• 优缺点
  ✅ 优势：存储空间占用小、关键词召回速度极快
  ❌ 缺点：无法识别同义词、近义词，语义模糊场景召回差

3. 实战最佳方案：稠密 + 稀疏「混合检索」

💡RAG 开发思路提炼： 稠密向量负责找「意思相近」的文档；稀疏向量负责找「关键词命中」的文档；两份召回结果经过重排融合，兼顾语义召回和关键词精准度，大幅度减少漏召回

二、代码实战 1：BGE‑M3模型 生成稠密 & 稀疏向量

BGE‑M3 是当下热门的多模态嵌入模型，可以一次性同时输出稠密向量 + 稀疏向量，无需分别调用两个模型。

""" 查看稠密向量和 稀疏向量 ->基于BGE-m3模型 """ from FlagEmbedding import BGEM3FlagModel #1.创建模型对象 model = BGEM3FlagModel("./assets/models/bge-m3") #2.对数据进行编码 result = model.encode( ["标量字段通常用来存储一些元数据，并可以在搜索时通过元数据进行过滤"], return_dense=True,# 是否返回稠密向量 return_sparse=True, # 是否返回稀疏向量 ) 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，字典，仅保存有意义关键词及其权重，长度不固定

三、代码实战 2：Milvus 创建向量数据表

创建 Milvus 表完整流程：连接 Milvus 实例 → 设计数据表 Schema 字段 → 创建向量索引 → 新建集合 (数据表)

数据表字段规划

完整建表代码

""" 创建向量数据库中的表 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", datatype=DataType.INT64,is_primary=True ).add_field( # 添加 text字段 类型为字符串,最大长度为1500 field_name="text",datatype=DataType.VARCHAR,max_length=1500 ).add_field( #添加metadata字段，类型为json field_name= "metadata",datatype=DataType.JSON ).add_field( #添加稠密向量字段，类型为浮点数向量，维度1024 field_name="vector",datatype=DataType.FLOAT_VECTOR,dim = 1024, ).add_field( #添加稀疏向量字段，类型为稀疏向量类型 field_name="sparse_vector",datatype=DataType.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_name="vector", # 建立索引的字段 index_type="HNSW", # 索引类型 metric_type="L2", # 向量相似度度量方式 ) index_params.add_index( #4.2稀疏向量索引 field_name="sparse_vector", index_type="SPARSE_INVERTED_INDEX", metric_type="IP", ) 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())

运行结果如图所示：

四、代码实战 3：Milvus添加向量数据

完整流程：加载本地 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( uri="http://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_size=400, # 每个块的最大长度 chunk_overlap=50, # 每个块重叠的长度 length_function=len, # 可选：计算文本长度的函数，默认为字符串长度，可自定义函数来实现按 token 数切分 add_start_index=True, # 可选：块的元数据中添加此块起始索引 ) #得到切分后的文档列表 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.使用嵌入模型切分的文档内容转为向量 #创建嵌入模型对象 model=BGEM3FlagModel("../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_name=collection_name,#关键字调用传参 data=data_list, #这里的data_list是data_list = [] ) # insert_data(get_milvus_client(),"demo_collection")

结果分析：

每一条文档切片，都会同时保存三份配套的数据：

1. metadata（元数据）
   记录这条文本来自哪个源文件路径，用于后续溯源文档来源。
2. vector（稠密向量）
   也就是稠密向量数组，负责语义检索，用来匹配意思相近的内容。
3. 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_name=collection_name, data=data_list )

循环里面的每一条字典数据，就对应可视化面板里一行入库记录

• metadata

  → 控制台metadata列

• vector

  → 控制台vector稠密向量列

• sparse_vector

  →控制台sparse_vector稀疏向量列

每一块切分出来的文档，都会生成一对一绑定的稠密向量与稀疏向量，同时存入向量数据库。

Milvus 数据添加执行流程

1. 首先调用get_milvus_client()，获取 Milvus 数据库的连接客户端；
2. 再传入集合名称demo_collection，两个参数一起传给insert_data入库函数；
3. 进入函数内部后，collection_name参数继续向下传递；
4. 接着组装每一条文档数据：稠密向量、稀疏向量、元数据、原文文本，打包成data_list；
5. 在client.insert()方法中，把collection_name和data_list传入；
6. 最终执行，将向量文档数据写入 Milvus 指定的集合中。

五、Milvus 删除向量数据

#从表中删除数据 from pymilvus import MilvusClient #1.获取连接数据库的的对象 def get_milvus_client(): #创建连接数据库的实例 client = MilvusClient( uri="http://127.0.0.1:19530", ) return client #2.删除数据的函数 def delete_data(client: MilvusClient,collection_name): #客户端对象去调用delete函数 client.delete( collection_name=collection_name,#表名 #过滤条件 filter="id in [468672494323370209]"#过滤条件:根据id来删 ) delete_data(get_milvus_client(),"demo_collection")

此记录468672494323370209已删除

六、代码实战 4：Milvus 检索｜单路检索 + 稠密稀疏混合检索

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( uri="http://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_dense=True, #是否返回稠密向量 return_sparse=True, #是否返回稀疏向量 ) #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_name=collection_name, #表名 data=[dense_vecs],#用户输入的查询数据转换成的稠密向量 anns_field="vector",#拿dense_vecs去查询数据库表中的哪一个字段->vector要和表中稠密向量的字段相同 limit=5,#代表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_name=collection_name, # 表名 data=[sparse_vecs], # 用户输入的查询数据转换成的 稀疏向量 anns_field="sparse_vector", # 拿sparse_vecs去查询数据库表中的哪一个字段 -> sparse_vector 稀疏向量 limit=5, # 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_field="vector", param = {"metric_type": "L2"}, limit=5 ) sparse_req = AnnSearchRequest( data=[sparse_vecs], anns_field="sparse_vector", param={"metric_type": "IP"}, limit=5 ) # 启动向量数据库的联合查询 results = client.hybrid_search( collection_name = collection_name, # 数据库表名 reqs = [dense_req, sparse_req], # 要查询内容的 稠密向量和稀疏向量 limit = 5, # top k ranker= RRFRanker(k=60), # 重排 output_fields = ["id","text","metadata"] # 需要查询的字段 ) return results #hybrid_search("监护人的职责是什么",get_milvus_client(),"demo_collection") #6.拿向量数据库中查询的结果 作为上下文和用户提示词提交给LLM->LLM生成最终的反馈内容 def main(query): #1.创建大语言模型对象 llm = ChatOpenAI( model_name="gpt-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 返回的检索结果里提取文档片段，拼接成给大模型使用的参考上下文。

我们逐层拆解：

1. results[0]

   ：取出混合检索命中的第一批最相关文档块；

2. for data in results[0]

   ：循环遍历每一条检索记录；

3. data["entity"]["text"]

   ：文本内容嵌套存放在entity对象内部，两层取值拿到原始文档片段。

⚠️避坑提醒：千万不要直接写data["text"]，外层字典没有 text 字段，直接取值代码就会报错。

1. "\n".join( )

   ：将多条命中的文档片段，用换行符隔开，合并成一整段文本；

2. 最后赋值给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%免费】