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})")