浅谈治理RAG 升级RAG 混合检索+重排序

发表于最近更新于最近

作者 Neo

14~19 分钟 阅读

前言

之前搭建的LangChain和LangGraph，使用的RAG有点不太准确，会出现幻觉，即答非所问，比如：

问"变速箱油多久换一次"，知识库里明明有答案，系统会回"建议参考用户手册"。

LLM 没在知识库里找到答案，但被要求必须回答，于是用通用知识硬编。

检索这块需要升级，打算采用混合检索和重排序的技术进行功能增强。

1、整改思路：

第一步：混合检索

向量检索擅长语义匹配。比如搜"OBD"，它能理解成"车载诊断系统"。但缺点也明显——如果知识库里写的是"OBD 接口位于方向盘下方"，其中的关键词"OBD"在语义上并不突出，向量可能把这条排到后面。

BM25 是关键词匹配的老牌算法，正好补这个缺。

所以做法很简单：向量和 BM25 两条路同时搜，结果合并。

第二步：RRF 融合

合并不是加权平均。向量返回的是余弦距离（0~1），BM25 返回的是词频统计分（可能是 0~20），量纲不同，直接加权没意义。

用 RRF（倒数排名融合），只看排名不看分数：

text

score = 1 / (60 + rank)

向量排第 1 的得 1/61，BM25 排第 1 的也得 1/61。公平。

第三步：重排序

粗召回 20 条之后，用 Cross-Encoder 再打一遍分，取 Top 3。

Bi-Encoder（向量检索）是海选，Cross-Encoder 是终面。前者把问题和文档分别编码，后者把问题和文档拼在一起让模型看，精度更高。

用的模型是 BAAI/bge-reranker-v2-m3。

2、代码升级：

“代码改动集中在三个文件：rag_service.py 负责检索逻辑，rag_tool.py 负责把检索挂到工具链上，agent_service.py 修了一个返回值的问题。”

节点网络：

rag_service.py

新增了四个方法：

python

def _build_bm25_index(self):
    all_data = self.vectorstore.get()
    self.doc_texts = all_data['documents']
    tokenized_corpus = [doc.split() for doc in self.doc_texts]
    self.bm25 = BM25Okapi(tokenized_corpus)

def hybrid_search(self, query: str, top_k: int = 20):
    vector_results = self.vectorstore.similarity_search_with_score(query, k=top_k)
    tokenized_query = query.split()
    bm25_scores = self.bm25.get_scores(tokenized_query)
    top_indices = np.argsort(bm25_scores)[-top_k:][::-1]
    return sorted_docs[:top_k]

async def _async_rerank(self, query: str, docs: List[Document], top_n: int = 3):
    loop = asyncio.get_event_loop()
    return await loop.run_in_executor(self._executor, self.reranker.rerank, query, docs, top_n)

async def advanced_retrieve(self, query: str) -> List[Document]:
    candidates = self.hybrid_search(query, top_k=20)
    return await self._async_rerank(query, candidates, top_n=3)

关键点：_async_rerank 用了 run_in_executor，把重排序扔到线程池里跑，不堵 FastAPI 的事件循环。

rag_tool.py

从自己维护 Chroma 连接改成调用 RAGService：

python

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

@tool
async def search_knowledge_base(query: str) -> str:
    rag = get_rag_service()
    docs = await rag.advanced_retrieve(query)
    return format_docs(docs)

agent_service.py

_build_graph 方法末尾：

python

return builder.compile()  # 之前漏了这一行

3、注意的地方：

模型路径指向错了

配置里 RERANKER_MODEL_PATH 指向了 bge-small-zh（Embedding 模型），不是 Reranker。

报错：

text

classifier.bias   | MISSING |
classifier.weight | MISSING |

sentence_transformers.CrossEncoder 加载时会检查分类器权重，Embedding 模型没有这层，自然报错。

解决：下载 bge-reranker-v2-m3，改路径。

清华源不好用

下载模型时一直报连接错误。

切阿里云镜像：

text

HF_ENDPOINT=https://hf-mirror.com

新版 huggingface-cli 已弃用，改用 hf：

bash

hf download BAAI/bge-reranker-v2-m3 --local-dir ./models/BAAI--bge-reranker-v2-m3

换模型要重建向量库

之前用 bge-small-zh（512 维），换了 bge-m3（1024 维）之后，旧向量和新模型维度不匹配，必须重建：

python

embedding_model = HuggingFaceEmbeddings(model_name="BAAI/bge-m3")
vectorstore = Chroma.from_documents(documents, embedding_model, persist_directory="./chroma_db")

4、整理：

数据测试，人工测了 10 个之前答错的问题，新版对了 9个。（“比如'刹车片多久换'之前答'视情况而定'，现在能从库里查到'前刹车片3-5万公里'。”）

性能的问题，响应延迟从 ~50ms 涨到 ~200ms（主要在重排序推理上），显存多了 1.3GB。因为我是用的荣耀本子，没有独立的英伟达显卡，但是好歹有Intel Arc 显卡，重新卸载了PyTorch,安装兼容的 PyTorch 和 IPEX，实现了XPU 加速，速度提了不少，但风扇转得有点猛（还好是笔记本电脑，底下装了个散热的外挂制冷风扇）。

RAG 不是"向量数据库 + LLM"就完了，检索质量才是天花板。

混合检索 + 重排序，原理不复杂，准确率从 ~60% 拉到了 ~90%，重点是写好agent_node和SYSTEM_PROMPT(agent_service中的核心)，调试通，当问题抛过来的时候，能启用自己写的tool工具类，然后在工具类rag_service把async_rerank和advanced_retrieve写明白了，就可以了。

万法归宗

许可协议: CC BY 4.0