手把手教你搭建生产级 RAG 问答系统：Milvus + BM25 + 多查询 + 重排序

大靠山

185人浏览 · 2026-08-20 21:55:19

大靠山 · 2026-08-20 21:55:19 发布

本文基于一个完整的实战项目，带你从零构建一个高精度的智能问答机器人。我们会使用 Milvus 向量数据库存储知识库，结合 BM25 关键词检索 和 向量检索 进行混合召回，再通过 多查询生成 和 FlashRank 重排序 提升答案准确率。所有代码都有详细注释，读完你就能理解现代 RAG 系统的核心设计思路。

环境准备

Python 版本

建议使用 Python 3.12

如果下载速度慢，可配置国内 PyPI 镜像。

项目背景：为什么要这么复杂？

一个简单的 RAG 流程是：用户提问 → 向量数据库检索相似段落 → 拼接上下文 → 大模型生成答案。但在实际落地中，你会遇到以下问题：

用户问法千奇百怪：同一个意图“文石哪款 pad 尺寸最大”，用户可能问“最大的电纸书是哪台”，向量检索可能召回不同结果。
向量相似 ≠ 语义相关：向量模型可能把“大尺寸”和“大电池”混淆，导致无关段落排在前面。
纯向量检索会漏掉专有名词、型号等精确关键词：比如“Tab13”这类词，向量模型可能不认识。

因此，我们设计了如下流水线：

多查询生成：让大模型根据原始问题生成 3 个不同角度的问法，分别去检索，扩大召回范围，这个我们上一篇也专门聊过。
混合召回：每条问法既走 Milvus 向量检索（语义相似），也走 BM25 关键词检索（字面匹配），合并候选文档。
重排序：用专门的轻量模型对所有候选段落按与原始问题的真实相关性重新打分，只保留最相关的前 2 条送入大模型，这个上一篇也用过。

最终，大模型只看到少量高质量上下文，答案自然更准确、更省钱。

完整代码与逐行解析

3.1 基础配置与镜像劫持

import os
import shutil
import time
os.environ["HF_ENDPOINT"] = "https://hf-mirror.com"
from dotenv import load_dotenv
from langchain_core.documents import Document
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_deepseek import ChatDeepSeek
from langchain_huggingface import HuggingFaceEmbeddings
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough, RunnableLambda
from langchain_core.output_parsers import StrOutputParser
from pymilvus import MilvusClient, DataType
from flashrank import Ranker, RerankRequest
from langchain_community.retrievers import BM25Retriever
import requests

作用：

HF_ENDPOINT 设置 HuggingFace 国内镜像，避免下载模型超时。
从 .env 文件加载 DEEPSEEK_API_KEY。
导入了 Milvus 客户端、FlashRank 重排序器、BM25 检索器等关键组件。

3.2 拦截 HuggingFace 下载请求（防止模型下载失败）

original_get = requests.get
def mirror_get(url, *args, **kwargs):
if "huggingface.co" in url:
url = url.replace("https://huggingface.co", "https://hf-mirror.com")
return original_get(url, *args, **kwargs)
requests.get = mirror_get

有些库（如 FlashRank）内部使用 requests 下载模型文件，此段代码强制替换为国内镜像，保证下载成功。

3.3 清理 Milvus 数据库文件（Windows 下常见占用问题）

DB_PATH = "milvus_demo.db"
COLLECTION_NAME = "docs"
def clean_milvus_db(db_path: str, retry_times=3):
# 先尝试关闭所有可能的连接
for _ in range(2):
try:
tmp_cli = MilvusClient(db_path)
tmp_cli.close()
del tmp_cli
except Exception:
pass
time.sleep(0.6)
# 删除 LOCK 文件（Windows 下常因进程未释放而残留）
lock_path = os.path.join(db_path, "LOCK")
if os.path.exists(lock_path):
try:
os.unlink(lock_path)
except Exception:
pass
# 循环删除整个数据库目录
for i in range(retry_times):
try:
if os.path.exists(db_path):
shutil.rmtree(db_path)
print(f"[清理成功] {db_path} 已删除")
return
except PermissionError as e:
print(f"[清理重试{i+1}] 文件占用: {e}")
time.sleep(1)
raise Exception("自动清理失败，执行这条命令手动清理：`Get-Process milvus-lite* -EA SilentlyContinue | Stop-Process -Force; Remove-Item milvus_demo.db -Recurse -Force`")

为什么需要这个函数？
Milvus Lite 在 Windows 下可能因为异常退出、多进程访问等原因导致数据库目录被占用，重新运行时会出现 PermissionError。这个函数尝试优雅关闭连接、删除锁文件，实在不行就抛出提示让用户手动清理。

3.4 读取文档并切分成小块

with open("knowledge.txt", "r", encoding="utf-8") as f:
text = f.read()
documents = [Document(page_content=text)]
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=300,          # 每块最多300字符
chunk_overlap=50,        # 重叠50字符，避免语义断裂
separators=["\n\n", "\n", "。", "！", "？", "，", " ", ""]
)
docs = text_splitter.split_documents(documents)
target_chunk_num = len(docs)   # 记录切片总数，用于后续判断数据库是否需要重建

切分策略：

块大小 300 字符，适合大部分段落；重叠 50 字符确保上下文连贯。
分隔符顺序从大到小，优先在段落、换行、句号处切分，保证语义完整性。

3.5 初始化向量化模型

embeddings = HuggingFaceEmbeddings(
model_name="BAAI/bge-large-zh-v1.5",
model_kwargs={"device": "cpu"},
encode_kwargs={"normalize_embeddings": True}
)
dim = len(embeddings.embed_query("test"))   # 获取向量维度

bge-large-zh-v1.5 是中文语义向量模型的优选，效果好、速度快。
normalize_embeddings=True 会将向量归一化，使得余弦相似度计算更稳定。

3.6 Milvus 初始化与数据写入

3.6.1 判断是否需要重建数据库

need_rebuild = True
try:
check_client = MilvusClient(DB_PATH)
if check_client.has_collection(COLLECTION_NAME):
stats = check_client.get_collection_stats(COLLECTION_NAME)
db_chunk_num = stats.get("row_count", 0)
if db_chunk_num == target_chunk_num:
need_rebuild = False
print(f"✅ 向量库匹配，复用现有数据｜库内切片:{db_chunk_num}")
else:
print(f"📌 切片数量不一致，执行重建｜库:{db_chunk_num} 本地:{target_chunk_num}")
check_client.close()
del check_client
except Exception as e:
print(f"📌 向量库读取异常，强制重建: {e}")
try:
check_client.close()
except:
pass
del check_client
time.sleep(0.5)

这段代码避免了重复写入。如果本地知识库没有变化（切片数量相同），就直接复用已有的 Milvus 数据库，加快启动速度。

3.6.2 创建集合并写入向量

client = None
if need_rebuild:
client = MilvusClient(DB_PATH)
# 定义表结构
schema = client.create_schema(auto_id=False, enable_dynamic_field=False)
schema.add_field(field_name="id", datatype=DataType.INT64, is_primary=True)
schema.add_field(field_name="vector", datatype=DataType.FLOAT_VECTOR, dim=dim)
schema.add_field(field_name="text", datatype=DataType.VARCHAR, max_length=2000)
# 创建索引：FLAT 索引，余弦相似度
index_params = client.prepare_index_params()
index_params.add_index(field_name="vector", index_type="FLAT", metric_type="COSINE")
client.create_collection(collection_name=COLLECTION_NAME, schema=schema, index_params=index_params)
# 批量写入向量
insert_data = []
for idx, doc in enumerate(docs):
vec = embeddings.embed_query(doc.page_content)
insert_data.append({"id": idx, "vector": vec, "text": doc.page_content})
client.insert(collection_name=COLLECTION_NAME, data=insert_data)
# 关键：将集合加载到内存，否则检索会报错
client.load_collection(COLLECTION_NAME)
print(f"✅ 向量库构建完成并加载，写入 {len(insert_data)} 条切片")
else:
client = MilvusClient(DB_PATH)
client.load_collection(COLLECTION_NAME)   # 复用库也要加载

解释：

id 是主键，手动赋值为 0,1,2,… 方便对应。
text 字段保存切片的原文，max_length=2000 足够容纳 300 字符的文本。
FLAT 索引是暴力搜索，适合小规模数据（几万条以内），准确率 100%；COSINE 相似度与归一化向量配合使用。
重要：Milvus Lite 需要显式调用 load_collection() 将数据加载到内存后才能执行 search，否则会报错。

3.7 带阈值过滤的 Milvus 检索函数

def milvus_retrieve(query, top_k=5, score_threshold=0.6):
query_vec = embeddings.embed_query(query)
results = client.search(
collection_name=COLLECTION_NAME,
data=[query_vec],
limit=top_k,
output_fields=["text"]
)
doc_list = []
for hit in results[0]:
if hit["distance"] >= score_threshold:   # distance 是余弦相似度，值越大越相似
doc_list.append(Document(page_content=hit["entity"]["text"]))
return doc_list

score_threshold=0.6 用来过滤掉相似度太低的片段，避免引入噪声。注意：由于向量已归一化，余弦相似度范围是 [-1,1]，0.6 表示中等以上相似度。
返回的仍是 LangChain 的 Document 对象，方便后续统一处理。

3.8 初始化大模型与重排序器

llm = ChatDeepSeek(model="deepseek-v4-pro", temperature=0)ranker = Ranker()

temperature=0 保证大模型输出确定、严谨。
Ranker() 默认使用 ms-marco-TinyBERT-L-2-v2 模型，也可通过参数指定其他模型。

3.9 多查询生成

query_gen_prompt = ChatPromptTemplate.from_template(
"为以下问题生成3个不同角度的相关中文问题，换行分隔。\n原问题：{question}\n相关问法："
)
query_gen_chain = query_gen_prompt | llm | StrOutputParser()

大模型会输出类似：

文石最大屏幕的电子书阅读器是哪款哪款文石平板尺寸最大文石电纸书尺寸对比

随后按换行分割成列表。

3.10 构建 BM25 检索器

bm25_retriever = BM25Retriever.from_documents(docs)  # docs 是所有切片文档

BM25 是经典的关键词检索算法，擅长精确匹配专有名词、型号等。我们基于同样的切片构建 BM25 索引，实现关键词层面的召回。

3.11 多查询 + BM25 混合召回

def multi_query_retrieve(question):
variants = query_gen_chain.invoke(question).split("\n")
all_docs = []
seen = set()   # 用于去重
q_list = [question] + [v.strip() for v in variants if v.strip()]
# 1. 对每个问法进行 Milvus 向量检索
for q in q_list:
chunk_list = milvus_retrieve(q, top_k=3)
for chunk in chunk_list:
cnt = chunk.page_content
if cnt not in seen:
seen.add(cnt)
all_docs.append(chunk)
# 2. BM25 关键词召回
bm25_docs = bm25_retriever.invoke(question)   # 默认 top_k=4
for doc in bm25_docs:
if doc.page_content not in seen:
seen.add(doc.page_content)
all_docs.append(doc)
return all_docs[:8]   # 限制候选池大小

流程说明：

原问题 + 3 个生成变体，共 4 个查询，每个查询用 Milvus 召回 top 3，理论上最多 12 条。
BM25 再召回 4 条，合并后去重。
最终限制最多 8 条候选文档进入重排序，避免候选太多导致重排过慢。

3.12 重排序，选出最相关的 2 条

def retrieve_multi_and_rerank(question):
raw_docs = multi_query_retrieve(question)
if not raw_docs:
return []
passages = [{"text": doc.page_content} for doc in raw_docs]
rerank_res = ranker.rerank(RerankRequest(query=question, passages=passages))
top_texts = [item["text"] for item in rerank_res[:2]]
return [d for d in raw_docs if d.page_content in top_texts]

FlashRank 会对每个候选段落与原始问题计算相关性分数，并按分数降序排列。我们只取前 2 条，确保送入大模型的上下文最相关。

3.13 组装 RAG 链并执行

template = """根据下面提供的上下文，回答问题。如果不知道答案，就说不知道。
上下文：
{context}
问题：{question}
回答："""
prompt = ChatPromptTemplate.from_template(template)
def format_docs(docs):
return "\n\n".join(doc.page_content for doc in docs)
rag_chain = (
{
"context": RunnableLambda(retrieve_multi_and_rerank) | RunnableLambda(format_docs),
"question": RunnablePassthrough()
}
| prompt
| llm
| StrOutputParser()
)
if __name__ == "__main__":
question = "文石哪款的pad尺寸最大？"
answer = rag_chain.invoke(question)
print("回答：", answer)
# 释放资源
client.release_collection(COLLECTION_NAME)
client.close()

LCEL 链执行顺序：

用户问题同时传给两个分支：上下文分支和问题分支。
上下文分支：retrieve_multi_and_rerank 获得最佳文档列表 → format_docs 拼成字符串。
将上下文和问题填入 Prompt，交给 DeepSeek 生成答案。
StrOutputParser 提取纯文本输出。

运行效果演示

假设 knowledge.txt 内容如下：

文石 Tab10 拥有一块 10.3 英寸的电子墨水屏，适合日常阅读。文石 Tab13 配备 13.3 英寸柔性屏幕，是目前文石尺寸最大的电纸书，适合阅读 PDF。文石 Nova Air 采用 7.8 英寸屏幕，主打便携。文石 Leaf 2 是 7 英寸，非常轻巧。

运行脚本后，控制台会输出类似：

📌 向量库读取异常，强制重建: ...✅ 向量库构建完成并加载，写入 4 条切片回答： 文石 Tab13 的 pad 尺寸最大，它配备了 13.3 英寸的屏幕。

常见问题与解决方案

5.1 Windows 下 Milvus 数据库文件被占用

症状：PermissionError: [WinError 32] 或 [WinError 5]。
解决：确保上次运行的 Python 进程已完全退出；或手动运行提示中的 PowerShell 命令强制结束 milvus-lite 进程并删除文件夹。

5.2 向量检索返回空列表

检查 score_threshold 是否设置过高，可尝试降低到 0.3。
确认 client.load_collection(COLLECTION_NAME) 是否已执行，否则 search 会报错或返回空。

5.3 重排序模型下载失败

请确保 requests.get 镜像劫持代码在 Ranker() 初始化之前已执行。
也可以手动下载模型并指定 cache_dir 参数，如：Ranker(model_name="ms-marco-TinyBERT-L-2-v2", cache_dir="你的路径")。

持久化向量库：Milvus Lite 让数据落盘，重启不丢失，也方便接入Milvus 。
混合召回：向量检索 + BM25 关键词检索，互补增强。
多查询生成：自动扩展用户意图，提高召回覆盖率。
重排序精排：FlashRank 轻量高效，筛选出真正相关的段落。
阈值过滤：过滤低相似度噪声，提升上下文质量。

接下来我们会逐步的封装成可对外使用的api。

学AI大模型的正确顺序，千万不要搞错了

🤔2026年AI风口已来！各行各业的AI渗透肉眼可见，超多公司要么转型做AI相关产品，要么高薪挖AI技术人才，机遇直接摆在眼前！

有往AI方向发展，或者本身有后端编程基础的朋友，直接冲AI大模型应用开发转岗超合适！

就算暂时不打算转岗，了解大模型、RAG、Prompt、Agent这些热门概念，能上手做简单项目，也绝对是求职加分王🔋

在这里插入图片描述

📝给大家整理了超全最新的AI大模型应用开发学习清单和资料，手把手帮你快速入门！👇👇

学习路线:

✅大模型基础认知—大模型核心原理、发展历程、主流模型（GPT、文心一言等）特点解析
✅核心技术模块—RAG检索增强生成、Prompt工程实战、Agent智能体开发逻辑
✅开发基础能力—Python进阶、API接口调用、大模型开发框架（LangChain等）实操
✅应用场景开发—智能问答系统、企业知识库、AIGC内容生成工具、行业定制化大模型应用
✅项目落地流程—需求拆解、技术选型、模型调优、测试上线、运维迭代
✅面试求职冲刺—岗位JD解析、简历AI项目包装、高频面试题汇总、模拟面经

以上6大模块，看似清晰好上手，实则每个部分都有扎实的核心内容需要吃透！

我把大模型的学习全流程已经整理📚好了！抓住AI时代风口，轻松解锁职业新可能，希望大家都能把握机遇，实现薪资/职业跃迁～