LLM - 使用 RAG (检索增强生成) 多路召回实现精准知识问答教程

欢迎关注我的CSDN：https://spike.blog.csdn.net/
本文地址：https://spike.blog.csdn.net/article/details/142629289

免责声明：本文来源于个人知识与公开资料，仅用于学术交流，欢迎讨论，不支持转载。

RAG

RAG (Retrieval-Augmented Generation，检索增强生成) 的多路召回，包括向量召回和本文召回，可用于精准知识问答，减轻大模型的幻觉问题，即：

并行：同时使用文本召回和向量召回，合计获得 TopN 个样本，再使用重排序的方式，获得 TopK 个样本，作为最终的召回文本。
串行：优先使用文本召回，召回 TopN 个样本，再使用向量排序，获得 TopK 个样本，作为最终的召回样本。

启动 Ollama 服务：

# 配置 HOST
export OLLAMA_HOST="0.0.0.0:11434"
# 配置 模型路径
export OLLAMA_MODELS="ollama_models"nohup ollama serve > nohup.ollama.out &

RAG 使用 LangChain 框架，参考：LangChain - Quickstart

LangChain 的相关依赖包，即：

pip install langchain
pip install beautifulsoup4
pip install faiss-cpu
pip install jiebapip install langchain-community
pip install langchain-huggingface
pip install rank_bm25
pip install langchain_openai

准备编码模型 BGE，即：

# https://huggingface.co/BAAI/bge-large-zh-v1.5
modelscope download --model BAAI/bge-large-zh-v1.5 --local_dir BAAI/bge-large-zh-v1.5

导入 LangChain 的相关 Python 包：

from typing import List
import jiebafrom langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.vectorstores import FAISS
from langchain.document_loaders import TextLoader
from langchain_huggingface import HuggingFaceEmbeddings
from langchain_community.retrievers import BM25Retriever

使用 LangChain 读取外部文档 medical_data.txt，即：

loader = TextLoader('medical_data.txt')
documents = loader.load()
text_splitter = RecursiveCharacterTextSplitter(chunk_size = 500,chunk_overlap  = 0,length_function = len,separators=['\n']
)
docs = text_splitter.split_documents(documents)

其中 medical_data.txt (4999 条) 格式如下，已经组织成 question 与 answer 的内容：

# ...
{'question': '曲匹地尔片的用法用量', 'answer': '注意：同种药品可由于不同的包装规格有不同的用法或用量。本文只供参考。如果不确定，请参看药品随带的说明书或向医生询问。口服。一次50～100mg（1-2片），3次/日，或遵医嘱。'}
# ...

Docs 是 list 格式，单项如下：

metadata 信息源
page_content 信息内容

即：

Document(metadata={'source': 'medical_data.txt'}, page_content="{'question': '曲匹地尔片的用法用量', 'answer': '注意：同种药品可由于不同的包装规格有不同的用法或用量。本文只供参考。如果不确定，请参看药品随带的说明书或向医生询问。口服。一次50～100mg（1-2片），3次/日，或遵医嘱。'}")

Query 是文档中已经问题，即：

query = "请问锁骨骨折多久能干活?"

使用 BM25Retriever 构建检索器，选择 TopK=10 个文档，因为是中文，预处理使用 Jieba 分词，即：

def preprocessing_func(text: str) -> List[str]:return list(jieba.cut(text))
retriever = BM25Retriever.from_documents(docs, preprocess_func=preprocessing_func, k=10)
bm25_res = retriever.invoke(query)

BM25 算法的核心，在于利用词频(Term Frequency, TF) 和逆文档频率(Inverse Document Frequency, IDF) 衡量文档与查询之间的相关性，同时引入文档长度信息，来调整相关性的计算。

构建向量 Embeddings 库：

embeddings = HuggingFaceEmbeddings(model_name='llm/BAAI/bge-large-zh-v1.5', model_kwargs = {'device': 'cuda:1'})
db = FAISS.from_documents(docs, embeddings)

其中 5000 条向量，构建 embeddings 需要 1min 15s，CPU 执行。

获取向量召回：

vector_res = db.similarity_search(query, k=10)

使用 RRF 算法，进行多路召回合并，10+10=20 选取最优的 10 个召回，即：

def rrf(vector_results: List[str], text_results: List[str], k: int=10, m: int=60):"""使用 RRF 算法对两组检索结果进行重排序params:vector_results (list): 向量召回的结果列表, 每个元素是专利IDtext_results (list): 文本召回的结果列表, 每个元素是专利IDk(int): 排序后返回前k个m (int): 超参数return:重排序后的结果列表,每个元素是(文档ID, 融合分数)"""doc_scores = {}# 遍历两组结果,计算每个文档的融合分数for rank, doc_id in enumerate(vector_results):doc_scores[doc_id] = doc_scores.get(doc_id, 0) + 1 / (rank+m)for rank, doc_id in enumerate(text_results):doc_scores[doc_id] = doc_scores.get(doc_id, 0) + 1 / (rank+m)# 将结果按融合分数排序sorted_results = [d for d, _ in sorted(doc_scores.items(), key=lambda x: x[1], reverse=True)[:k]]return sorted_resultsvector_results = [i.page_content for i in vector_res]
text_results = [i.page_content for i in bm25_res]
rrf_res = rrf(vector_results, text_results)

RRF (Reciprocal Rank Fusion, 倒数排名融合) 算法将多个检索结果合并一个聚合列表，通过每个列表中每个项目的排名取倒数，即 1 除以排名，将倒数排名在所有列表中相加，得到每个项目的最终得分。

提示词工程：

prompt = '''
任务目标：根据检索出的文档回答用户问题
任务要求：1、不得脱离检索出的文档回答问题2、若检索出的文档不包含用户问题的答案，请回答我不知道用户问题：
{}检索出的文档：
{}
'''

使用 Ollama 服务进行大模型推理，注意需要使用长 Token 模型，即：

from langchain_community.llms import Ollama
model = Ollama(model="qwen-2_5-32b-max-context:latest")
print(f"[Info] rrf_res: {len(rrf_res)}")
full_prompt = prompt.format(query, ''.join(rrf_res))
# print(f"[Info] prompt: {full_prompt}")
res = model.invoke(full_prompt)  # RAG
print(f"[Info] response: {res}")res = model.invoke(query)  # 非 RAG
print(f"[Info] response: {res}")

RAG 的输出，与文档高度一致，即：

锁骨骨折的恢复时间一般在3个月左右。虽然骨折刚刚愈合时可以进行轻微的工作，但若涉及重体力劳动，则通常需要大约半年的时间才能重新开始，最少也需要4-5个月。过早地从事重体力工作有可能导致骨折处再次受伤。因此，在这期间避免过度负重活动是十分重要的，以确保锁骨能完全恢复并维持愈合效果。

非 RAG 的输出：

锁骨骨折的恢复时间取决于骨折的严重程度以及治疗方法。一般来说，轻微到中度的锁骨骨折可能需要大约6-8周的时间来初步愈合，在这段时间内，患者可能会被建议限制肩部和上肢的活动以促进骨折部位的稳定与修复。
但是，能否重新开始工作还依赖于具体工作的性质。如果工作不需要使用受伤的手臂或肩膀进行高强度劳动，则在几周后可能就可以慢慢恢复工作。然而，如果是需要手臂大力操作的工作，则可能需要等待3个月甚至更长时间才能安全地返回工作岗位，并且最好等到医生确认骨折完全愈合为止。
因此，在考虑重返岗位之前，应该咨询主治医师的意见，确保不会对康复过程造成负面影响或导致二次伤害。

参考：https://github.com/wyf3/llm_related

全部源码：

from typing import Listimport jieba
from langchain.document_loaders import TextLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.llms import Ollama
from langchain_community.retrievers import BM25Retriever
from langchain_community.vectorstores import FAISS
from langchain_huggingface import HuggingFaceEmbeddingsclass RagRetriever(object):"""RAG retriever"""def __init__(self):loader = TextLoader(db_path)documents = loader.load()text_splitter = RecursiveCharacterTextSplitter(chunk_size=500,chunk_overlap=0,length_function=len,separators=['\n'])docs = text_splitter.split_documents(documents)def preprocessing_func(text: str) -> List[str]:return list(jieba.cut(text))self.doc_retriever = BM25Retriever.from_documents(docs, preprocess_func=preprocessing_func, k=10)print("[Info] init doc done!")embeddings = HuggingFaceEmbeddings(model_name=bge_path,model_kwargs={'device': 'cuda:1'})self.db = FAISS.from_documents(docs, embeddings)print("[Info] init db done!")self.prompt = '''任务目标：根据检索出的文档回答用户问题任务要求：1、不得脱离检索出的文档回答问题2、若检索出的文档不包含用户问题的答案，请回答我不知道用户问题：{}检索出的文档：{}'''print("[Info] init all done!")@staticmethoddef rrf(vector_results: List[str], text_results: List[str], k: int = 10, m: int = 60):"""使用 RRF 算法对两组检索结果进行重排序params:vector_results (list): 向量召回的结果列表, 每个元素是专利IDtext_results (list): 文本召回的结果列表, 每个元素是专利IDk(int): 排序后返回前k个m (int): 超参数return:重排序后的结果列表,每个元素是(文档ID, 融合分数)"""doc_scores = {}# 遍历两组结果,计算每个文档的融合分数for rank, doc_id in enumerate(vector_results):doc_scores[doc_id] = doc_scores.get(doc_id, 0) + 1 / (rank + m)for rank, doc_id in enumerate(text_results):doc_scores[doc_id] = doc_scores.get(doc_id, 0) + 1 / (rank + m)# 将结果按融合分数排序sorted_results = [d for d, _ in sorted(doc_scores.items(), key=lambda x: x[1], reverse=True)[:k]]return sorted_resultsdef retrieve(self, query):bm25_res = self.doc_retriever.invoke(query)vector_res = self.db.similarity_search(query, k=10)vector_results = [i.page_content for i in vector_res]text_results = [i.page_content for i in bm25_res]rrf_res = self.rrf(vector_results, text_results)model = Ollama(model="qwen-2_5-32b-max-context:latest")print(f"[Info] rrf_res: {len(rrf_res)}")full_prompt = self.prompt.format(query, ''.join(rrf_res))# print(f"[Info] prompt: {full_prompt}")res1 = model.invoke(full_prompt)print(f"[Info] rag response: {res1}")res2 = model.invoke(query)print(f"[Info] n-rag response: {res2}")return res1, res2def main():query = "请问锁骨骨折多久能干活?"rr = RagRetriever()rr.retrieve(query)if __name__ == '__main__':main()