Reranker + BM25 + FAISS 构建高效的多阶段知识库检索系统一

一、什么是知识库检索？

在构建基于大语言模型的问答系统（如 RAG）中，知识库检索（Retrieval） 是第一步，也是影响最终回答质量的关键环节。它负责从大规模文档中快速定位与用户问题最相关的 top-k 段落。下面提供的是一个思路方向，包括代码的大概实现步骤。

本文介绍一种经典的三段式检索流程：

BM25：粗排召回
FAISS：向量相似度排序
Reranker：精排打分

二、为什么使用多阶段检索？

阶段	技术	功能
第一阶段	BM25	基于关键词匹配，召回相关文档
第二阶段	FAISS	基于语义向量，进行高效近似最近邻搜索
第三阶段	Reranker	基于交叉注意力机制，精准排序候选结果

这种“先快后准”的策略既保证了效率又提升了准确性。

三、实现思路概述

[Query] ↓
BM25 → [Top-50 Candidates]↓
FAISS → [Top-10 Semantically Similar]↓
Reranker → [Top-3 最佳匹配段落]

四、环境依赖安装

pip install rank_bm25 faiss-cpu sentence-transformers torch transformers

五、第一阶段：BM25 召回

from rank_bm25 import BM25Okapi
import jieba  # 中文分词示例# 模拟知识库
corpus = ["大模型训练需要大量数据和高性能计算资源","RAG 系统通过外部知识提升回答能力","Faiss 是 Facebook 开发的高效向量检索库","BM25 是一个基于统计的语言模型"
]tokenized_corpus = [list(jieba.cut(doc)) for doc in corpus]
bm25 = BM25Okapi(tokenized_corpus)# 查询
query = list(jieba.cut("如何提升问答系统的准确性？"))
top_n_docs = bm25.get_top_n(query, corpus, n=3)print("BM25召回结果：")
for doc in top_n_docs:print(" -", doc)

六、第二阶段：FAISS 向量检索

from sentence_transformers import SentenceTransformer
import faiss
import numpy as np# 加载预训练语义编码器
model = SentenceTransformer('paraphrase-MiniLM-L6-v2')# 编码文档
doc_embeddings = model.encode(corpus)
query_embedding = model.encode(["如何提升问答系统的准确性？"])# 构建 FAISS 索引
dimension = doc_embeddings.shape[1]
index = faiss.IndexFlatL2(dimension)
index.add(np.array(doc_embeddings))# 检索 top-3
_, indices = index.search(np.array(query_embedding), k=3)
faiss_results = [corpus[i] for i in indices[0]]print("FAISS语义检索结果：")
for doc in faiss_results:print(" -", doc)

七、第三阶段：Reranker 精排打分

from transformers import AutoTokenizer, AutoModelForSequenceClassification
import torch# 加载 reranker 模型（中文）
tokenizer = AutoTokenizer.from_pretrained("BAAI/bge-reranker-base")
model = AutoModelForSequenceClassification.from_pretrained("BAAI/bge-reranker-base")def rerank(query, candidates):pairs = [[query, doc] for doc in candidates]inputs = tokenizer(pairs, padding=True, truncation=True, return_tensors='pt')with torch.no_grad():scores = model(**inputs).logits.squeeze().cpu().numpy()return sorted(zip(candidates, scores), key=lambda x: x[1], reverse=True)reranked = rerank("如何提升问答系统的准确性？", faiss_results)print("Reranker 排序结果：")
for doc, score in reranked:print(f" - {doc} (Score: {score:.2f})")

八、完整流程整合

你可以将上述三个阶段串联成完整的检索 pipeline：

def multi_stage_retrieval(query, corpus, bm25_top_k=10, faiss_top_k=5, rerank_top_k=3):# 1. BM25召回tokenized_query = list(jieba.cut(query))bm25_candidates = bm25.get_top_n(tokenized_query, corpus, n=bm25_top_k)# 2. FAISS语义排序query_emb = model.encode([query])doc_embs = model.encode(bm25_candidates)_, indices = index.search(query_emb.reshape(1, -1), k=faiss_top_k)faiss_candidates = [bm25_candidates[i] for i in indices[0]]# 3. Reranker 打分reranked = rerank(query, faiss_candidates)return reranked[:rerank_top_k]# 调用
result = multi_stage_retrieval("如何提升问答系统的准确性？", corpus)
for doc, score in result:print(f"✅ {doc} (Rerank Score: {score:.2f})")

九、总结

方法	优点	缺点
BM25	快速、无需训练	不支持语义理解
FAISS	支持高维向量检索	无法处理复杂语义关系
Reranker	精准排序	计算开销略大

✅ 推荐组合使用：BM25 + FAISS + Reranker，既能保证效率又能提升准确率。

十、结语

本文介绍了如何结合 BM25、FAISS 和 Reranker 实现一个高效的多阶段知识库检索系统。这种方案非常适合用于本地化 RAG 应用、企业级问答系统或智能客服平台。

📌 欢迎点赞、收藏，并关注我，我会持续更新更多关于 AI、LLM、RAG、向量数据库等内容！

十一、导出 Markdown 文件

以下是完整的 .md 格式内容，请你复制保存为 knowledge_retrieval_guide.md 即可用于发布或归档。

# 🔍 使用 Reranker + BM25 + FAISS 构建高效的多阶段知识库检索系统## 一、什么是知识库检索？在构建基于大语言模型的问答系统（如 RAG）中，**知识库检索（Retrieval）** 是第一步，也是影响最终回答质量的关键环节。它负责从大规模文档中快速定位与用户问题最相关的 top-k 段落。本文介绍一种经典的三段式检索流程：1. **BM25：粗排召回**
2. **FAISS：向量相似度排序**
3. **Reranker：精排打分**---## 二、为什么使用多阶段检索？| 阶段 | 技术 | 功能 |
|------|------|------|
| 第一阶段 | BM25 | 基于关键词匹配，召回相关文档 |
| 第二阶段 | FAISS | 基于语义向量，进行高效近似最近邻搜索 |
| 第三阶段 | Reranker | 基于交叉注意力机制，精准排序候选结果 |这种“先快后准”的策略既保证了效率又提升了准确性。---## 三、实现思路概述

[Query]
↓
BM25 → [Top-50 Candidates]
↓
FAISS → [Top-10 Semantically Similar]
↓
Reranker → [Top-3 最佳匹配段落]


---## 四、环境依赖安装```bash
pip install rank_bm25 faiss-cpu sentence-transformers torch transformers

五、第一阶段：BM25 召回

from rank_bm25 import BM25Okapi
import jieba# 模拟知识库
corpus = ["大模型训练需要大量数据和高性能计算资源","RAG 系统通过外部知识提升回答能力","Faiss 是 Facebook 开发的高效向量检索库","BM25 是一个基于统计的语言模型"
]tokenized_corpus = [list(jieba.cut(doc)) for doc in corpus]
bm25 = BM25Okapi(tokenized_corpus)# 查询
query = list(jieba.cut("如何提升问答系统的准确性？"))
top_n_docs = bm25.get_top_n(query, corpus, n=3)print("BM25召回结果：")
for doc in top_n_docs:print(" -", doc)

六、第二阶段：FAISS 向量检索

from sentence_transformers import SentenceTransformer
import faiss
import numpy as np# 加载预训练语义编码器
model = SentenceTransformer('paraphrase-MiniLM-L6-v2')# 编码文档
doc_embeddings = model.encode(corpus)
query_embedding = model.encode(["如何提升问答系统的准确性？"])# 构建 FAISS 索引
dimension = doc_embeddings.shape[1]
index = faiss.IndexFlatL2(dimension)
index.add(np.array(doc_embeddings))# 检索 top-3
_, indices = index.search(np.array(query_embedding), k=3)
faiss_results = [corpus[i] for i in indices[0]]print("FAISS语义检索结果：")
for doc in faiss_results:print(" -", doc)

七、第三阶段：Reranker 精排打分

from transformers import AutoTokenizer, AutoModelForSequenceClassification
import torch# 加载 reranker 模型（中文）
tokenizer = AutoTokenizer.from_pretrained("BAAI/bge-reranker-base")
model = AutoModelForSequenceClassification.from_pretrained("BAAI/bge-reranker-base")def rerank(query, candidates):pairs = [[query, doc] for doc in candidates]inputs = tokenizer(pairs, padding=True, truncation=True, return_tensors='pt')with torch.no_grad():scores = model(**inputs).logits.squeeze().cpu().numpy()return sorted(zip(candidates, scores), key=lambda x: x[1], reverse=True)reranked = rerank("如何提升问答系统的准确性？", faiss_results)print("Reranker 排序结果：")
for doc, score in reranked:print(f" - {doc} (Score: {score:.2f})")

八、完整流程整合

你可以将上述三个阶段串联成完整的检索 pipeline：

def multi_stage_retrieval(query, corpus, bm25_top_k=10, faiss_top_k=5, rerank_top_k=3):# 1. BM25召回tokenized_query = list(jieba.cut(query))bm25_candidates = bm25.get_top_n(tokenized_query, corpus, n=bm25_top_k)# 2. FAISS语义排序query_emb = model.encode([query])doc_embs = model.encode(bm25_candidates)_, indices = index.search(query_emb.reshape(1, -1), k=faiss_top_k)faiss_candidates = [bm25_candidates[i] for i in indices[0]]# 3. Reranker 打分reranked = rerank(query, faiss_candidates)return reranked[:rerank_top_k]# 调用
result = multi_stage_retrieval("如何提升问答系统的准确性？", corpus)
for doc, score in result:print(f"✅ {doc} (Rerank Score: {score:.2f})")