RAG检索增强生成是当前大模型知识库问答的主流落地方式,它把外部文档检索结果注入提示词,再由大模型生成回答,能有效缓解幻觉问题。本文按问答链路拆解RAG检索增强生成的工程步骤,包括文本切分、向量化、多路召回与重排,并给出可直接运行的Python代码。
RAG检索增强生成的架构与适用场景
RAG检索增强生成的标准流程包含五个环节:文档解析、文本切分、向量入库、检索召回、生成回答。相比直接微调模型,RAG的数据更新成本低,新增文档只需重新入库,不用改动模型权重;代价是回答质量受检索结果影响大,检索不到正确片段时模型仍然会编造内容。
适用场景集中在三类:企业内部知识库问答、产品文档检索、私有化资料查询。对推理能力要求高、且答案完全依赖模型知识的任务(如代码生成、数学推导)不适合套RAG,直接调用大模型更合适。
文本切分与向量化入库
切分质量决定检索上限。按固定字符硬切会把段落上下文切断,建议按标题层级递归切分,chunk_size控制在500到1000字之间,overlap设为100到150字,保留上下文连续性。
from langchain_community.document_loaders import TextLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_community.embeddings import HuggingFaceEmbeddings
from langchain_community.vectorstores import Chroma
loader = TextLoader("kb/network_ops.txt", encoding="utf-8")
docs = loader.load()
splitter = RecursiveCharacterTextSplitter(
chunk_size=600,
chunk_overlap=120,
separators=["\n## ", "\n### ", "\n\n", "\n", "。", " "]
)
chunks = splitter.split_documents(docs)
embeddings = HuggingFaceEmbeddings(model_name="BAAI/bge-large-zh-v1.5")
vectorstore = Chroma.from_documents(chunks, embeddings, persist_directory="./kb_db")
print(f"已写入 {len(chunks)} 个向量块")
embedding模型建议选中文优化的bge系列,检索效果普遍好于通用多语言向量模型。入库时为每个文档块记录来源、更新时间和权限标签,方便检索时过滤过期内容。
向量检索与关键词多路召回
单一向量检索在专业术语场景召回率偏低,常规做法是向量检索与BM25关键词检索并行,再把结果融合排序。向量路负责语义相似,BM25路负责精确词匹配,融合后取分数最高的前若干条作为生成上下文。
from rank_bm25 import BM25Okapi
import jieba
def bm25_scores(query, corpus):
tokenized = [list(jieba.cut(c)) for c in corpus]
bm25 = BM25Okapi(tokenized)
return bm25.get_scores(list(jieba.cut(query)))
def rrf_fusion(vector_rank, bm25_rank, k=60):
fused = {}
for rank in (vector_rank, bm25_rank):
for i, doc_id in enumerate(rank[:20]):
fused[doc_id] = fused.get(doc_id, 0) + 1.0 / (k + i + 1)
return sorted(fused.items(), key=lambda x: x[1], reverse=True)
融合后的top结果控制在5到8条,上下文过长会稀释提示词信号,过长过短都会影响回答完整度。
重排模型提升排序准确率
召回阶段得到的顺序侧重词面相关,重排阶段用reranker对query与候选片段逐对打分,把真正相关的段落提到最前。bge-reranker-v2-m3在中文场景表现稳定,推理成本低于一次模型生成。
from FlagEmbedding import FlagReranker
reranker = FlagReranker("BAAI/bge-reranker-v2-m3")
pairs = [[query, chunk] for chunk in candidates]
scores = reranker.compute_score(pairs, normalize=True)
order = sorted(range(len(candidates)), key=lambda i: scores[i], reverse=True)
final_chunks = [candidates[i] for i in order[:5]]
生成阶段提示词模板
生成提示词要同时约束两件事:只依据检索内容回答,检索不到时明确说不知道。两条约束能显著降低幻觉率。
PROMPT = """你是一个知识库问答助手。只依据下面检索到的资料回答,不要使用资料之外的知识。
资料:
{context}
问题:{question}
回答规则:
1. 资料中没有的信息,直接回答"知识库中未收录该信息";
2. 引用资料时标注片段编号;
3. 用中文回答。"""
print(PROMPT.format(context=ctx, question=query))
调优评估指标
用真实问题构建评测集,统计三项指标:检索命中率(检索结果是否包含答案)、回答准确率(生成结果与人工答案一致的比例)、无答案率。检索命中率低于80%时优先调整切分策略和embedding模型,不要急着改生成提示词;命中率达标后再逐步收紧回答规则。
原创文章,作者:小编,如若转载,请注明出处:https://www.yunthe.com/rag-jian-suo-zeng-qiang-sheng-cheng-shi-zhan-xiang-liang/