浅谈治理RAG 混合检索+重排序 (续)
前言
在使用普通检索（倒排索引/BM25）的项目里，检索算法本身是看字不看意，导致命中率不高，给出的结果也不够准确，这个时候需要升级两种技术，混合检索、重排序。
先说一下混合检索、重排序：
混合检索：BM25关键词检索+向量检索+RRF公平合并（由看字变为理解意思在找）
重排序：根据原文深层理解后重新排序（根据原文更深入理解意思重打一遍分）
混合检索用到的场景比较多、至于重排序要根据自己的硬件情况决定采不采用（比较吃性能）
1、一些model的简介和准备：
BAAI--bge-small-zh-v1.5、BAAI--bge-m3、bge-reranker-v2-m3
BAAI/bge-small-zh-v1.5：轻量级文本嵌入模型、向量维度512
BAAI/bge-m3:：全能型多语言文本嵌入模型、向量维度1024
bge-reranker-v2-m3：重排序模型
下载好上面的模型，原先有v1.5模型，新下载bge-m3和reranker-v2-m3，下面会用到
使用最新的bge-m3升级向量数据库：
# ...之前逻辑不动...
#初始化模型
embedding_model = HuggingFaceEmbeddings(
model_name="./models/BAAI--bge-m3",
# bge-m3 模型在检索任务中需要归一化
model_kwargs={'device': 'xpu'}, # 如果你有GPU，可以改为 'cuda'
encode_kwargs={'normalize_embeddings': True}
)
#生成m3向量库
vectorstore = Chroma.from_documents(
documents=documents,
embedding=embedding_model,
persist_directory="./chroma_db_m3"
)
# ...后面不变...升级的原因是为了使用向量维度1024的向量库，配合混合检索+重排序提升准确率（暂不考虑硬件消耗）
2、配置文件：
使用混合检索和重排，需要修改Config、把reranker-v2-m3模型和一些参数写入配置：
from pydantic_settings import BaseSettings
class Settings(BaseSettings):
PROJECT_NAME: str = "LangChain FastAPI Assistant"
OLLAMA_MODEL: str = "qwen2.5:3b"
OLLAMA_BASE_URL: str = "http://localhost:11434"
# 混合检索参数
HYBRID_TOP_K: int = 20
RERANK_TOP_N: int = 3
RRF_K: int = 60
# 重排序模型
RERANKER_MODEL_PATH: str = "./models/bge-reranker-v2-m3"
class Config:
env_file = ".env"
settings = Settings()2、服务层RAG：
在rag_service要加入混合检索和重排序的核心方法：
import os
os.environ["HF_ENDPOINT"] = "https://hf-mirror.com"
import json
import asyncio
from concurrent.futures import ThreadPoolExecutor
from typing import AsyncGenerator, List
import numpy as np
from rank_bm25 import BM25Okapi
from langchain_community.vectorstores import Chroma
from langchain_huggingface import HuggingFaceEmbeddings
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_core.documents import Document
from app.core.llm import get_llm
from app.core.reranker import LocalReranker
from app.core.config import settings
def format_docs(docs: list[Document]) -> str:
"""辅助函数：将检索出来的多个 Document 拼接成一段长文本"""
formatted = []
for i, doc in enumerate(docs, 1):
formatted.append(f"[参考资料 {i}]（分类: {doc.metadata.get('category')}）:\n{doc.page_content}")
return "\n\n".join(formatted)
class RAGService:
def __init__(self):
# 1. 加载相同的 Embedding 模型
self.embedding_model = HuggingFaceEmbeddings(
model_name="./models/BAAI--bge-m3", #使用bge-m3
#我电脑是Intel Arc 显卡、可以使用xpu加速，如果没有去掉默认cpu
model_kwargs={'device': 'xpu'},
encode_kwargs={'normalize_embeddings': True}
)
# 2. 读取本地持久化的向量库
self.vectorstore = Chroma(
persist_directory="./chroma_db_m3",
embedding_function=self.embedding_model
)
# 3. 转换为检索器 (k=2 代表检索最相似的前 2 条 FAQ 记录) —— 保留原样
self.retriever = self.vectorstore.as_retriever(search_kwargs={"k": 2})
# 4. 获取 LLM
self.llm = get_llm()
# 5. 专为 RAG 设计的 Prompt 模板
self.prompt = ChatPromptTemplate.from_template(
"""你是一个专业且严谨的汽车售后智能客服助手。请严格根据下面提供的【参考知识库】回答用户的问题。
规则要求：
1. 如果【参考知识库】中有相关信息，请结合参考内容，用规范、礼貌的客服口吻回答。
2. 如果【参考知识库】中完全没有涉及该问题，请明确告知用户：“抱歉，关于该问题目前知识库中暂未收录，建议您拨打官方热线或来店进行实车检测。”，不要凭空捏造。
【参考知识库】：
{context}
【用户问题】：
{question}
"""
)
# 使用 LCEL 串联 RAG 管道
self.chain = (
{
"context": self.retriever | format_docs,
"question": RunnablePassthrough()
}
| self.prompt
| self.llm
)
#RAG
# 重排序器
self.reranker = LocalReranker(model_name=settings.RERANKER_MODEL_PATH)
#
self._executor = ThreadPoolExecutor(max_workers=2)
# 构建 BM25 索引
self._build_bm25_index()
# ---------- [NEW] 构建 BM25 索引 ----------
def _build_bm25_index(self):
"""从现有 Chroma 向量库中取出所有文档，构建 BM25 索引"""
all_data = self.vectorstore.get()
# all_data 包含 'ids', 'documents', 'metadatas'
self.doc_texts = all_data['documents'] # list of str
self.doc_ids = all_data['ids'] # list of str
if not self.doc_texts:
self.bm25 = None
return
# 分词
tokenized_corpus = [doc.split() for doc in self.doc_texts]
self.bm25 = BM25Okapi(tokenized_corpus)
#混合检索、向量、BM25、RRF
def hybrid_search(self, query: str, top_k: int = None) -> List[Document]:
"""
返回经过 RRF 融合后的候选文档列表（最多 top_k 条）
每个 doc 的 metadata 中会添加 'hybrid_score' 字段
"""
if top_k is None:
top_k = settings.HYBRID_TOP_K
if not self.doc_texts:
return []
# 1) 向量检索
vector_results = self.vectorstore.similarity_search_with_score(query, k=top_k)
vec_docs = []
for doc, dist in vector_results:
doc.metadata["_vec_rank"] = len(vec_docs)
doc.metadata["_vec_score"] = 1 - dist
vec_docs.append(doc)
#BM25 检索
bm25_docs = []
if self.bm25 and self.doc_texts:
tokenized_query = query.split()
scores = self.bm25.get_scores(tokenized_query)
top_indices = np.argsort(scores)[-top_k:][::-1]
for rank, idx in enumerate(top_indices):
doc = Document(
page_content=self.doc_texts[idx],
metadata={
"id": self.doc_ids[idx],
"_bm25_rank": rank,
"_bm25_score": scores[idx],
}
)
bm25_docs.append(doc)
#RRF
fused_scores = {}
def get_doc_key(doc):
return doc.page_content
#先初始化所有参与文档的分数为0
all_candidate_docs = vec_docs + bm25_docs
for doc in all_candidate_docs:
key = get_doc_key(doc)
if key not in fused_scores:
fused_scores[key] = {
'doc': doc,
'score': 0.0,
'metadata': doc.metadata.copy()
}
#对向量排名加分
for doc in vec_docs:
rank = doc.metadata.get("_vec_rank", top_k)
fused_scores[get_doc_key(doc)]['score'] += 1 / (settings.RRF_K + rank + 1)
for doc in bm25_docs:
rank = doc.metadata.get("_bm25_rank", top_k)
fused_scores[get_doc_key(doc)]['score'] += 1 / (settings.RRF_K + rank + 1)
sorted_items = sorted(fused_scores.values(), key=lambda x: x['score'], reverse=True)
result_docs = []
for item in sorted_items[:top_k]:
doc = item['doc']
doc.metadata['hybrid_score'] = item['score']
for key in ['_vec_rank', '_vec_score', '_bm25_rank', '_bm25_score']:
doc.metadata.pop(key, None)
result_docs.append(doc)
return result_docs
# 异步重排序
async def _async_rerank(self, query: str, docs: List[Document], top_n: int = None) -> List[Document]:
"""在独立线程中执行重排序，避免阻塞异步事件循环"""
if top_n is None:
top_n = settings.RERANK_TOP_N
if not docs:
return []
loop = asyncio.get_event_loop()
# 使用 run_in_executor 执行同步的 rerank 方法
reranked = await loop.run_in_executor(
self._executor,
self.reranker.rerank,
query,
docs,
top_n
)
return reranked
#高级检索
async def advanced_retrieve(self, query: str) -> List[Document]:
"""完整的流程：混合检索 -> 重排序，返回最终精排后的文档列表"""
#混合检索
candidates = self.hybrid_search(query, top_k=settings.HYBRID_TOP_K)
if not candidates:
return []
#重排序
final_docs = await self._async_rerank(query, candidates, top_n=settings.RERANK_TOP_N)
return final_docs
#返回结果
async def get_stream_response(self, question: str) -> AsyncGenerator[str, None]:
"""流式返回 RAG 生成的结果（使用原有简单检索）"""
async for chunk in self.chain.astream(question):
if chunk.content:
payload = json.dumps({"content": chunk.content}, ensure_ascii=False)
yield f"data: {payload}\n\n"3、工具层调用RAG：
from langchain_core.tools import tool
from app.services.rag_service import RAGService
#单例模式
_rag_service = None
def get_rag_service():
global _rag_service
if _rag_service is None:
_rag_service = RAGService()
return _rag_service
def format_docs(docs) -> str:
"""格式化检索到的文档，显示置信度得分"""
if not docs:
return "未找到相关内容。"
formatted = []
for i, doc in enumerate(docs, 1):
score = doc.metadata.get("rerank_score", doc.metadata.get("hybrid_score", 0.0))
formatted.append(f"[参考资料 {i}] (置信度: {score:.4f}):\n{doc.page_content}")
return "\n\n".join(formatted)
@tool
async def search_knowledge_base(query: str) -> str:
"""
异步检索私有知识库（汽车FAQ），内置混合检索 + 重排序精刷。
当用户询问汽车相关问题（如保养、故障、配置等）时调用此工具。
"""
rag = get_rag_service()
docs = await rag.advanced_retrieve(query)
if not docs:
return "知识库中未检索到相关内容，请尝试其他问题或联系人工客服。"
top_score = docs[0].metadata.get("rerank_score", -100)
if top_score < -2.0:
return "未找到高度匹配的内容，建议咨询官方渠道。"
return format_docs(docs)4、整理：
在AgentService里，有两个地方要注意：
1、写好SYSTEM_PROMPT，确认当触发与知识库相关的内容、【必须】调用工具检索知识库。（为了更好触发自己写的tool类）
2、写好agent_node，进入节点后，提前写好一个KEYWORDS，然后检测messages是不是触发到关键词，如果触发到，就调用自己写的search_knowledge_base。（也是为了更加触发自己写的tool类）
AgentService能改的地方不多，主要写好prompt和node就可以了，代码不在粘了。
本章作为上一篇的姊妹篇，更底层更直白的讲述了RAG治理的核心，也展示出了最核心的混合检索和重排序代码，关于RAG暂时先告一段落。
许可协议:
CC BY 4.0