RAG实战：从文档加载到API封装的知识库问答系统

这次我们来看一个 RAG 检索增强生成问答系统的完整实战。重点不是重复“RAG 是什么”的概念，而是把整条链路跑起来：文档加载、中文分块、BM25 稀疏检索、稠密向量检索、RRF 倒数排名融合、Prompt 拼接、调用大模型生成答案，最后封装成 API 和批量任务。整篇文章会带代码，适合已经了解 RAG 基本概念、想亲手实现一个最小知识库问答系统的读者。

这套方案的硬件门槛并不高。检索和融合环节主要吃 CPU 和内存，不需要单独的大显存；嵌入模型可以选小尺寸中文模型；生成环节可以接本地大模型（比如 Ollama 服务的 Qwen、DeepSeek），也可以接 OpenAI 兼容接口。换句话说，即使没有独立显卡，把生成模型换成远端 API，整套流程依然可以跑通。

文章会绕开 LangChain 这种重量级框架，先用最直观的方式把底层逻辑拆明白，确认每一步都理解之后，再考虑要不要迁移到成熟框架。最后给出批量问答和 FastAPI 接口封装，方便直接接入真实业务。

1. RAG 核心能力速览

这里要强调一句：RAG 项目没有固定公式，不同场景对分块粒度、检索路数、融合策略、生成模型的要求都不一样。本文给的是最小可运行基线，后面所有参数都可以按实际数据调整。

2. RAG 整体架构与关键环节

RAG 的完整流程可以拆成四个阶段：数据准备、索引构建、检索召回、生成回答。下面这条链路是当前工业界比较通用的形态。

知识库文档 -> 加载解析 -> 文本分块 -> 向量化与索引构建 | 用户问题 -> 查询短语处理 -> 双路检索 --------| v RRF 融合排序 | Prompt 拼接 | 大模型生成 | 最终答案

拆开看每个环节的职责：

• 文档加载：把 PDF、Word、Markdown、TXT 等非结构化数据转成纯文本。这一步容易出问题的是 PDF 排版错乱、表格被拆散、页眉页脚混入正文。
• 文本分块：把长文档切成适合检索和模型输入的片段。分块太短会丢失上下文，太长会引入噪声，还容易超出模型上下文窗口。
• 向量化与索引：将文本片段转换成向量。这里有两种常见路线，一种是稠密向量，用深度模型把文本映射成固定维度向量；另一种是稀疏向量，用 BM25、TF-IDF 这类基于词频统计的方法。
• 检索召回：用户提问后，从知识库中找到最相关的若干片段。
• 融合排序：多路检索结果合并，最常用的方案之一就是 RRF 倒数排名融合。
• Prompt 拼接：把检索片段和用户问题组织成结构化提示词。
• 大模型生成：将完整 Prompt 输入大模型，输出答案。

这个流程看着不算长，但每一层都有参数和工程细节。下面从环境准备开始，逐步实现。

3. 环境准备与依赖安装

3.1 创建虚拟环境

建议使用独立虚拟环境，避免项目依赖污染系统 Python。

python -m venv rag-demo # Windows rag-demo\Scripts\activate # Linux / macOS source rag-demo/bin/activate

3.2 安装 Python 依赖

核心依赖包含分词的 jieba、BM25 实现的 rank_bm25、向量化编码的 sentence-transformers、数值计算的 numpy，以及用于 API 调用的 requests。

pip install jieba rank_bm25 sentence-transformers numpy requests

如果还要做 API 服务，再安装 FastAPI 相关依赖。

pip install fastapi uvicorn pydantic

如果涉及 PDF 解析，额外安装 pypdf。

pip install pypdf

需要特别注意的是 sentence-transformers 会连带安装 PyTorch。如果你的机器没有配置好 CUDA，建议先安装 CPU 版 PyTorch，避免自动下载一个很大的 CUDA 依赖。

pip install torch --index-url https://download.pytorch.org/whl/cpu pip install sentence-transformers

有独立显卡且已装好 CUDA 的机器，这一步可以跳过，直接让 pip 自动匹配 GPU 版本。

4. 文档加载与分块策略

数据清洗和分块是 RAG 系统里最容易被低估的一环。很多人把时间花在调大模型接口上，结果检索召回一堆废话，问题往往就出在文档没有处理好。

4.1 文本文件加载

先用一个简单的文本读取函数做通用模板。真实项目中，路径和编码需要按实际情况调整。

def load_txt(path: str) -> str: with open(path, "r", encoding="utf-8") as f: return f.read()

4.2 PDF 文件加载

PDF 解析比纯文本麻烦一些，常见坑是扫描版 PDF 需要 OCR、表格被错误换行、多栏排版读取顺序错乱。先用 pypdf 做最基础的解析。

from pypdf import PdfReader def load_pdf(path: str) -> str: reader = PdfReader(path) pages = [] for page in reader.pages: text = page.extract_text() if text: pages.append(text) return "\n".join(pages)

如果你的知识库主要是扫描件，这个方案不够，需要接 OCR 服务；如果以 Word、HTML 为主，可以使用 python-docx、BeautifulSoup 等工具。替换读取函数即可。

4.3 中文文本分块

分块策略直接影响检索效果。常见方案有三种。

第一，固定窗口分块，简单粗暴但容易切断语义；第二，递归分隔符分块，按标题、段落、句子的优先级逐级切分，LangChain 的 RecursiveCharacterTextSplitter 就是这个思路；第三，语义分块，用嵌入模型判断句子边界，效果更好但代价更高。

给出一个适合中文的轻量分块方案：先按句号、问号、感叹号、分号切句，再按最大长度合并，同时保留重叠片段，减少切分带来的信息丢失。

import re def split_into_sentences(text: str): parts = re.split(r"(?<=[。！？；])", text) return [p.strip() for p in parts if p.strip()] def split_into_chunks(text: str, chunk_size: int = 300, overlap: int = 50): sentences = split_into_sentences(text) chunks = [] buffer = "" for sent in sentences: if len(buffer) + len(sent) >= chunk_size: if buffer: chunks.append(buffer) if overlap > 0: buffer = buffer[-overlap:] + sent else: buffer = sent else: buffer += sent if buffer: chunks.append(buffer) return chunks

这里 chunk_size 和 overlap 都不是固定值。一般先用 200 到 500 字做实验，看检索召回效果再调整。如果文档有明确的标题结构，优先按标题层级切分，再对每个章节做句子级切分，效果通常会更好。

5. 双路检索：BM25 稀疏检索 + 稠密向量检索

RAG 的检索阶段通常不会只依赖一条路。纯稠密向量检索在语义理解上很擅长，但在精确匹配人名、编号、设备型号、特殊参数时，表现不如关键词检索；纯 BM25 又没有办法处理同义改写情况。所以更稳的方案是双路检索，最后用融合算法合并结果。

先准备一小批演示文档，模拟一个知识库。

documents = [ "RAG（Retrieval-Augmented Generation）检索增强生成，通过在生成前检索外部知识库，把相关内容作为上下文补充给大模型。", "稀疏向量检索的代表算法是 BM25。BM25 基于词频和逆文档频率对文档打分，适合精确关键词匹配。", "稠密向量检索使用深度神经网络将文本映射为固定维度向量，通过余弦相似度衡量语义相关性。", "RRF（Reciprocal Rank Fusion）倒数排名融合算法，输入多路检索结果，输出按融合分数排序的最终结果列表。", "常见 RAG 框架包括 LangChain、LlamaIndex、Dify 等。Dify 提供可视化工作流，适合快速搭建知识库应用。", "中文文档分块建议优先按段落、标题、句子边界切分，避免把一个完整语义单元拆散。", "大模型幻觉问题可以通过 RAG 引入外部事实来缓解，前提是检索结果要准确、上下文要完整。", "向量数据库常见选型包括 FAISS、Milvus、Chroma、Qdrant。FAISS 轻量，适合本地实验。", "构建 RAG 知识库时，文档质量直接影响检索效果。清洗格式、去除无关页眉页脚是关键步骤。", "评估 RAG 系统通常关注检索召回率、上下文相关性、答案准确率和回答可溯源四个维度。", ]

5.1 BM25 稀疏检索实现

BM25 的核心思想是：词在文档中出现得越多，文档得分越高；但这个词如果在整个文档集合中出现得越频繁，它的权重就要下调。简单说，既看重词频，又惩罚普遍出现的词。

代码使用 rank_bm25 库，中文先做 jieba 分词。

import jieba import numpy as np from rank_bm25 import BM25Okapi tokenized_docs = [list(jieba.cut(doc)) for doc in documents] bm25 = BM25Okapi(tokenized_docs) def sparse_search(query: str, top_k: int = 3): query_tokens = list(jieba.cut(query)) scores = bm25.get_scores(query_tokens) top_indices = np.argsort(scores)[::-1][:top_k] return [(int(idx), float(scores[idx])) for idx in top_indices]

返回结果是不定长数组，因为np.argsort(scores)[::-1]在 scores 全部相同时会给出从大到小的索引。top_indices转换成 int 后可以直接用于 documents 列表索引。

5.2 稠密向量检索实现

稠密向量这部分选一个对中文友好的小尺寸嵌入模型。BAAI/bge-small-zh-v1.5 是值得先试的模型，体积小，语义表现稳定。首次运行会自动下载模型文件，下载时间取决于网络条件。

from sentence_transformers import SentenceTransformer embedder = SentenceTransformer("BAAI/bge-small-zh-v1.5") doc_vecs = embedder.encode(documents, normalize_embeddings=True) def dense_search(query: str, top_k: int = 3): query_vec = embedder.encode(query, normalize_embeddings=True) sims = np.dot(doc_vecs, query_vec) top_indices = np.argsort(sims)[::-1][:top_k] return [(int(idx), float(sims[idx])) for idx in top_indices]

注意normalize_embeddings=True后，向量点积就等于余弦相似度，后续不需要再手动计算余弦公式，直接 np.dot 即可。

5.3 对比两种检索效果

可以自己跑几组查询对比感受。

查询“BM25 和稠密向量有什么区别”，BM25 更容易命中带“BM25”“稠密向量”字样的片段，稠密向量检索则可能召回“语义相关性”“向量化”相关的片段；查询“如何评估 RAG 系统”，BM25 可能因“评估”一词精准命中，而稠密向量也能通过语义关联找到“评估 RAG 系统”相关片段。

两条路的结果不完全一样，这正是需要融合的原因。

6. RRF 倒数排名融合算法原理与实现

6.1 RRF 核心公式

RRF 的全称是 Reciprocal Rank Fusion，倒数排名融合。它的核心思想不依赖具体分数，而是只依赖排名位置。公式如下。

score(d) = Σ 1 / ( k + rank_i(d) )

其中：

• rank_i(d) 表示文档 d 在第 i 路检索结果中的排名，从 0 开始计；
• k 是平滑常数，原论文中通常取 60；
• Σ 表示对多路检索结果求和。

使用排名而不是原始分值，好处很明显：不同检索器输出的分数尺度可能完全不同，向量相似度可能是 0.6 到 0.9，BM25 分数可能是几到几十，直接加权平均很不公平。RRF 把每路结果统一成“第几名”，用排名参与计算，天然规避了分数尺度不一致的问题。

6.2 RRF 代码实现

实现非常短。

def rrf_fusion(ranked_lists, k: int = 60): fused = {} for ranked in ranked_lists: for rank, doc_id in enumerate(ranked): fused[doc_id] = fused.get(doc_id, 0) + 1.0 / (k + rank + 1) return sorted(fused.items(), key=lambda x: x[1], reverse=True)

enumerate(ranked)从 0 开始，所以分母用k + rank + 1。如果某个文档只在其中一路出现，它只获得这一路的分数；两路都出现且排名靠前的文档，融合分数会明显更高。

6.3 融合效果验证

用一个例子手动算一遍。

假设某文档 A 在稀疏检索中排第 1 名，在稠密检索中排第 3 名，k 取 60。

A 的融合分数 = 1/(60+1) + 1/(60+3) = 0.01639 + 0.01587 = 0.03226

另一篇文档 B 在稀疏检索中排第 2 名，在稠密检索中排第 1 名。

B 的融合分数 = 1/(60+2) + 1/(60+1) = 0.01613 + 0.01639 = 0.03252

可以看到 B 的融合分数略高，因为它拿到了一个“第 1 名”和一个“第 2 名”，整体排名质量稍好。

把双路检索接在一起：

def hybrid_search(query: str, top_k: int = 3): sparse_top = sparse_search(query, top_k=5) dense_top = dense_search(query, top_k=5) sparse_ids = [doc_id for doc_id, _ in sparse_top] dense_ids = [doc_id for doc_id, _ in dense_top] fused = rrf_fusion([sparse_ids, dense_ids]) return fused[:top_k]

这里检索时先各取 5 条，再融合取前 3 条。之所以多取一些再截断，是因为融合阶段有时会出现“单路排名第 6 但两路都出现”的文档，综合排序可能比“单路第 3”更靠前。

7. 大模型生成与完整问答流程

检索到相关片段后，下一步就是组装 Prompt 并调用大模型。

7.1 Prompt 设计

RAG 的 Prompt 设计核心是三点：明确角色、提供资料、限制编造。

SYSTEM_PROMPT = "你是一个严谨的知识库问答助手。请根据提供的资料回答问题。如果资料中没有相关信息，直接说不知道，不要编造。" def build_prompt(question: str, context_chunks) -> str: context = "\n\n".join( [f"[片段{i+1}] {documents[doc_id]}" for i, (doc_id, _) in enumerate(context_chunks)] ) return f"资料：\n{context}\n\n问题：{question}\n\n请基于资料回答："

context_chunks是hybrid_search返回的结果，每个元素是(doc_id, score)元组。这里把文档原文拼进 Prompt，并保留片段编号，方便后续做答案溯源。

7.2 调用大模型

大模型接入方式有很多种。本地推荐用 Ollama，它启动后提供 OpenAI 兼容接口。这里以 Ollama 为例，模型名需要按本机实际拉取的模型替换。

import requests LLM_BASE_URL = "http://127.0.0.1:11434/v1" LLM_MODEL = "qwen2.5:7b-instruct" LLM_API_KEY = "ollama" def generate_answer(question: str, context_chunks, temperature: float = 0.3) -> str: prompt = build_prompt(question, context_chunks) resp = requests.post( f"{LLM_BASE_URL}/chat/completions", headers={"Authorization": f"Bearer {LLM_API_KEY}"}, json={ "model": LLM_MODEL, "messages": [ {"role": "system", "content": SYSTEM_PROMPT}, {"role": "user", "content": prompt}, ], "temperature": temperature, "max_tokens": 512, }, timeout=120, ) resp.raise_for_status() data = resp.json() return data["choices"][0]["message"]["content"]

如果本机 Ollama 版本不支持 OpenAI 兼容端点，也可以调用原生/api/chat接口，把LLM_BASE_URL换成http://127.0.0.1:11434/api/chat，参数格式略有不同。接云端大模型时，只需要改成对应的接口地址、密钥和模型名。

7.3 完整问答脚本

把前面的临时环境变量和函数整合成一个单文件脚本，方便跑通全流程。这个脚本只是演示，正式项目里建议把索引构建和查询服务拆成两个模块。

import jieba import numpy as np import requests from rank_bm25 import BM25Okapi from sentence_transformers import SentenceTransformer documents = [...] # 上一节定义的演示文档 # BM25 索引 tokenized_docs = [list(jieba.cut(doc)) for doc in documents] bm25 = BM25Okapi(tokenized_docs) # 稠密向量索引 embedder = SentenceTransformer("BAAI/bge-small-zh-v1.5") doc_vecs = embedder.encode(documents, normalize_embeddings=True) def sparse_search(query, top_k=5): tokens = list(jieba.cut(query)) scores = bm25.get_scores(t