你好，我是老周。在做知识库类项目时，经常会出现一个尴尬场景：同一个 RAG 系统，换了部门文档后效果天差地别；甚至同批文档，换一种切片策略，回答质量就不一样。调参调到最后，大家都不知道问题出在文档解析、切片、召回还是重排环节。这篇文章我就围绕“模块化 RAG 项目”这个主题，从架构设计、代码实现、评估指标和工程落地四个维度做一个完整拆解。不管你是刚接触 RAG 的新人，还是已经在做知识库落地的后端开发，都可以照着这套思路去设计和改造自己的项目。
1. 什么是模块化 RAG，为什么我们需要它
1.1 从传统 RAG 到模块化 RAG
RAG（Retrieval-Augmented Generation，检索增强生成）是一种把检索系统和生成式大模型结合的技术方案。它的核心思路是：不直接把用户问题丢给大模型，而是先从知识库中检索出相关文档片段，再把“用户问题 + 检索片段”一起组合成 Prompt，交给大模型生成答案。这样做的最大好处是，模型可以从外部知识库获取实时或私有知识，减少幻觉，也能把知识源的更新成本从“重新训练模型”降为“更新索引”。
早期的 RAG 项目通常是一个线性流水线：
文档加载 -> 文本切分 -> 向量化 -> 存入向量库 -> 检索 -> 拼接 Prompt -> 大模型生成这种流程在 Demo 阶段跑得很顺，但在真实业务中会遇到一系列问题：
- 文档类型多样，PDF、Word、Markdown、HTML 混合在一起，单一加载器处理不了。
- 表格、图片、图表中的信息经常被忽略或乱码。
- 切片大小对检索效果影响明显，但很难找到一组“通用参数”。
- 向量检索结果噪声大，相关文档往往排在第五名之后。
- 用户一次提问涉及多个知识点，单路检索返回内容不全。
这些问题都指向一个核心诉求：RAG 系统需要拆开做、按模块调优，于是就有了“模块化 RAG”。
1.2 模块化 RAG 的核心设计思想
模块化 RAG 并不是一种新技术，而是一种工程组织方式。它把 RAG 流水线中的每一个环节抽象成独立模块，模块之间通过标准接口通信，每个模块可以单独替换、单独测试、单独部署。
模块化 RAG 的关键设计原则可以概括为：
- 单一职责：每个模块只做一件事。例如文档解析模块只负责把 PDF 转成文本和结构化数据，不掺入向量化的逻辑。
- 接口标准化：模块之间通过统一的数据结构传递。在 Python 里，通常使用尽量通用的对象（例如文档对象、切片对象）而不是绑定框架的数据类型。
- 可插拔：同一个业务场景可以自由组合不同模块。比如今天用 OpenAI Embedding，明天换国产模型，不应该影响上游切片和下游检索。
- 可观测：每个模块运行完毕都要有日志、指标和中间产物，方便定位是哪一环出了问题。
要注意，模块化 RAG 和“用 LangChain Flow”不是一个概念。LangChain 等框架提供了模块化编排能力，但真正决定项目工程质量的是你自己的模块边界划分，以及每个模块内部的实现质量。
2. 模块化 RAG 整体架构与模块划分
2.1 九个核心模块
一个可落地的模块化 RAG 项目，通常由以下九个模块组成。我按数据处理顺序排列：
| 模块名称 | 职责 | 输入 | 输出 |
|---|---|---|---|
| 数据接入模块 | 从本地、数据库、OSS、API 拉取原始文件 | 原始文件字节流 | 文件列表 |
| 文档解析模块 | 将 PDF、Word、HTML 等转为纯文本和结构数据 | 文件 | 文档对象 |
| 文本切分模块 | 按语义或长度规则切分文本 | 文档对象 | 切片对象 |
| 向量化模块 | 将文本转为 Embedding 向量 | 切片对象 | 向量数据 |
| 索引存储模块 | 写入向量库，建立倒排或 HNSW 索引 | 向量数据 | 索引记录 |
| 查询理解模块 | 处理用户问题，可能涉及改写、扩写、多路召回 | 用户问题 | 查询列表 |
| 检索模块 | 从向量库和关键词索引中召回候选文档 | 查询向量 | 候选文档列表 |
| 重排模块 | 对候选文档做精排，去除噪声 | 候选文档列表 | 精排结果 |
| 生成模块 | 组装 Prompt，调用大模型生成答案并附带引用 | 用户问题 + 精排文档 | 答案与引用信息 |
这里需要强调的是，“查询理解”是模块化 RAG 比传统 RAG 多出来的重要一环。传统 RAG 直接把用户原问题拿去向量化检索，但真实用户问题往往是缺主语的短句，例如“审批流程是什么”。如果知识库中有多个流程文档，纯向量召回效果就很不稳定。查询理解模块可以先做意图识别、问题补全，甚至通过反问澄清用户需求。
2.2 模块间数据流转
模块化 RAG 的数据流不是一条简单的直线，而是带有反馈回路的网络：
原始文件 -> 数据接入 -> 文档解析 -> 文本切分 -> 向量化 -> 索引存储 ↑ 用户问题 -> 查询理解 -> 检索模块 -> 候选文档 -> 重排 -> 生成模块 -> 答案 ↑ | └----- 反馈评估 ---------┘索引一侧是离线流程，查询一侧是在线流程。离线流程负责把知识库变成可检索的索引，在线流程负责实时处理用户请求。离线流程可以做成定时任务或事件触发，在线流程需要保证低延迟。
反馈回路是模块化 RAG 的一个重要优势。我们可以记录每次用户提问、检索结果、重排结果和最终答案，用这些数据去评估各模块效果，再反向优化切片策略、检索策略和 Prompt 模板。
3. 环境准备与项目结构
3.1 运行环境
模块化 RAG 的代码实现可以使用多种语言，但目前生态最成熟的是 Python。本文示例基于 Python 环境，重点把模块化思想和核心逻辑讲清楚，不依赖任何特定框架。你可以根据自己的项目技术栈把思路迁移到 Java、Go 或 Node.js。
版本方面需要根据你的实际环境调整，这里只列出建议：
- Python 3.10 或更高版本。
- 向量数据库可以使用 Chroma（本地开发）、FAISS（轻量检索）、Milvus 或 Qdrant（生产环境）。
- Embedding 模型可以使用 OpenAI 的 text-embedding-3-small，也可以使用开源的 bge-m3、bge-large-zh、m3e-base 等中文本地模型。
- 大模型接口可以使用 OpenAI 兼容协议，也可以使用国内大模型或本地部署模型。
- 文档解析建议配合 PyMuPDF、python-docx、BeautifulSoup 等库使用。
如果你的生产环境要求私有化部署，Embedding 模型和生成模型都需要切换为本地模型，代码层要做一层模型封装。
3.2 项目目录结构
一个模块化 RAG 项目的目录结构，我建议按照模块边界来组织，而不是按照“controller/service/mapper”这种传统后端结构来组织。下面是一个经过整理的参考结构：
rag-project/ ├── app/ │ ├── __init__.py │ ├── main.py # 入口，负责组装流程 │ ├── config/ │ │ ├── __init__.py │ │ └── settings.py # 全局配置 │ ├── ingestion/ # 离线数据接入与索引 │ │ ├── __init__.py │ │ ├── loader.py # 数据接入模块 │ │ ├── parser.py # 文档解析模块 │ │ ├── splitter.py # 文本切分模块 │ │ ├── embedder.py # 向量化模块 │ │ └── indexer.py # 索引存储模块 │ ├── retrieval/ # 在线查询与检索 │ │ ├── __init__.py │ │ ├── query.py # 查询理解模块 │ │ ├── retriever.py # 检索模块 │ │ ├── reranker.py # 重排模块 │ │ └── generator.py # 生成模块 │ ├── schema/ │ │ ├── __init__.py │ │ └── models.py # 统一数据结构 │ └── utils/ │ ├── __init__.py │ └── logger.py # 日志工具 ├── data/ │ ├── raw/ # 原始文件 │ └── processed/ # 中间产物 ├── tests/ │ ├── test_loader.py │ ├── test_splitter.py │ └── test_retriever.py ├── requirements.txt └── README.md这种目录的好处非常明显：新同学接项目时，一眼就能看出每个模块的入口在哪里，出了问题能快速定位到对应文件，而不是在几百行的 Service 里翻逻辑。
3.3 基础依赖
如果使用 Python，可以基于以下依赖构建：
pymupdf>=1.23.0 python-docx>=1.1.0 beautifulsoup4>=4.12.0 langchain-text-splitters>=0.2.0 chromadb>=0.4.0 sentence-transformers>=2.2.0 openai>=1.0.0注意：langchain-text-splitters是 LangChain 生态中非常值得单独复用的一部分，它只包含文本切分器，不包含完整的链式调用逻辑。把它作为模块化项目的一个组件使用，比直接依赖整个 LangChain 更轻量。
4. 核心模块拆解与代码实战
这一节是全文重点。我会按照离线索引和在线检索两条链路，把每个核心模块的关键代码写出来。为了便于理解，本文示例不直接使用 LangChain 的链式 API，而是尽量用纯 Python 实现核心逻辑，突出模块化设计本身。
4.1 统一数据结构
模块化 RAG 的前提是模块之间使用统一的中间数据结构。我建议定义一个轻量的文档对象和切片对象：
# 文件路径：app/schema/models.py from dataclasses import dataclass, field from typing import Optional @dataclass class Document: doc_id: str # 文档唯一 ID title: str # 文档标题 content: str # 文档正文内容 metadata: dict = field(default_factory=dict) # 来源、作者、时间等元信息 @dataclass class Chunk: chunk_id: str # 切片唯一 ID doc_id: str # 所属文档 ID content: str # 切片文本内容 metadata: dict = field(default_factory=dict) # 页码、章节等定位信息 @dataclass class RetrievalResult: chunk: Chunk score: float # 召回相关性得分 source: str # 来源模块，用于区分向量检索、关键词检索等这里之所以用dataclass而不是直接使用 LangChain 的Document对象，是为了降低模块间对框架的依赖。你可以很容易地把这个对象转换成任何框架需要的格式。
4.2 数据接入与文档解析模块
文档解析是 RAG 项目中最容易被低估的一环。很多团队做一个 Demo 只处理干净的 Markdown 文本，一旦进入真实业务，面对扫描版 PDF、客户发来的加密 Word、页面结构复杂的 HTML，解析质量立刻塌方。
文档解析模块的建议实现方式：
# 文件路径：app/ingestion/parser.py import io import fitz # PyMuPDF from bs4 import BeautifulSoup from docx import Document as DocxDocument from app.schema.models import Document class DocumentParser: """将不同格式的原始文件解析为 Document 对象。""" def parse(self, file_name: str, file_bytes: bytes) -> Document: ext = file_name.rsplit(".", 1)[-1].lower() if ext == "pdf": return self._parse_pdf(file_name, file_bytes) elif ext == "docx": return self._parse_docx(file_name, file_bytes) elif ext in ("html", "htm"): return self._parse_html(file_name, file_bytes) elif ext in ("md", "txt"): return self._parse_text(file_name, file_bytes) else: raise ValueError(f"Unsupported file type: {ext}") def _parse_pdf(self, file_name: str, file_bytes: bytes) -> Document: doc = fitz.open(stream=file_bytes, filetype="pdf") content_parts = [] metadata = {"page_count": len(doc), "source": file_name} for page_num, page in enumerate(doc, start=1): text = page.get_text("text") content_parts.append(f"\n=== 第{page_num}页 ===\n{text}") content = "\n".join(content_parts) return Document(doc_id=file_name, title=file_name, content=content, metadata=metadata) def _parse_docx(self, file_name: str, file_bytes: bytes) -> Document: doc = DocxDocument(io.BytesIO(file_bytes)) paragraphs = [p.text for p in doc.paragraphs if p.text.strip()] content = "\n".join(paragraphs) return Document(doc_id=file_name, title=file_name, content=content, metadata={"source": file_name}) def _parse_html(self, file_name: str, file_bytes: bytes) -> Document: soup = BeautifulSoup(file_bytes.decode("utf-8", errors="ignore"), "html.parser") title = soup.title.string.strip() if soup.title else file_name content = soup.get_text(separator="\n", strip=True) return Document(doc_id=file_name, title=title, content=content, metadata={"source": file_name}) def _parse_text(self, file_name: str, file_bytes: bytes) -> Document: content = file_bytes.decode("utf-8", errors="ignore") return Document(doc_id=file_name, title=file_name, content=content, metadata={"source": file_name})这段代码有几个地方值得注意：
- 解析结果中保留页码信息，便于最终答案生成时定位引用。
- 解析失败时抛出明确的异常，而不是静默返回空文档。
- HTML 解析时把标题计入元信息，便于检索后展示“来自哪个页面”。
4.3 文本切分模块
文本切分直接决定召回效果。切得太大，检索到的片段包含大量无关内容，干扰生成；切得太小，语义不完整，检索经常漏掉关键信息。
我建议的切分策略是“按语义边界优先，兼顾长度约束”。具体来说，优先按标题、段落、句子边界切分，如果段落过长，再按窗口截断。
# 文件路径：app/ingestion/splitter.py import re import uuid from app.schema.models import Chunk, Document class TextSplitter: """基于结构和长度规则的文本切分器。""" def __init__(self, chunk_size: int = 500, chunk_overlap: int = 80): self.chunk_size = chunk_size self.chunk_overlap = chunk_overlap def split(self, document: Document) -> list[Chunk]: # 第 1 步：按双换行拆分为段落，再从段落中统计句子 paragraphs = [p.strip() for p in re.split(r"\n\s*\n", document.content) if p.strip()] chunks = [] current_buffer = "" current_start = 0 metadata = dict(document.metadata) for para in paragraphs: # 如果当前缓冲加上新段落超过阈值，就先把当前缓冲切出去 if len(current_buffer) + len(para) > self.chunk_size and current_buffer: chunks.extend(self._cut_long_text(current_buffer, document.doc_id, metadata)) # 保留重叠部分 current_buffer = current_buffer[-self.chunk_overlap:] if self.chunk_overlap > 0 else "" current_buffer += "\n" + para if current_buffer else para if current_buffer: chunks.extend(self._cut_long_text(current_buffer, document.doc_id, metadata)) return chunks def _cut_long_text(self, text: str, doc_id: str, metadata: dict) -> list[Chunk]: """处理超长段落，按窗口滑动切分。""" chunks = [] start = 0 while start < len(text): end = min(start + self.chunk_size, len(text)) chunk_text = text[start:end] chunk_id = uuid.uuid4().hex chunk_meta = dict(metadata) chunk_meta["char_start"] = start chunk_meta["char_end"] = end chunks.append(Chunk(chunk_id=chunk_id, doc_id=doc_id, content=chunk_text, metadata=chunk_meta)) if end == len(text): break start = end - self.chunk_overlap return chunks这里需要解释几个关键参数：
chunk_size：单个切片的目标字符数。中文场景下 300 到 800 都是常见范围。过小会丢失语义，过大容易引入噪声。chunk_overlap：相邻切片之间的重叠字符数，目的是保留上下文衔接信息。- 我的实现里给每个切片保存了
char_start和char_end，这能帮助在回答问题时回跳原文。
实际项目中你可以做“切片策略配置化”，即把切片参数放进配置中心，针对不同业务域使用不同参数。这也是模块化的优势。
4.4 Embedding 与索引模块
Embedding 模块将所有文本切片转为向量表示。现阶段做中文 RAG，我建议优先测试开源模型 bge-m3 或 bge-large-zh，它们在中文语义和长文本上的表现比较稳定。
向量化模块封装：
# 文件路径：app/ingestion/embedder.py from sentence_transformers import SentenceTransformer class LocalEmbedder: """基于本地模型实现文本向量化，避免外部 API 依赖。""" def __init__(self, model_name: str = "BAAI/bge-m3"): self.model = SentenceTransformer(model_name, device="cpu") def embed_texts(self, texts: list[str]) -> list[list[float]]: # normalize_embeddings=True 会做向量归一化，便于内积近似余弦相似度 vectors = self.model.encode( texts, normalize_embeddings=True, show_progress_bar=False ) return vectors.tolist()这里要注意，如果使用 OpenAI 等 API 服务，要把 Embedding 模块接口统一抽象为embed_texts，这样切换模型时不需要修改上游代码。生产环境建议把模型加载放到独立进程中，避免每次请求都重新加载模型。
索引模块负责把向量写入向量数据库。这里以 Chroma 为例：
# 文件路径：app/ingestion/indexer.py import chromadb from app.schema.models import Chunk class ChromaIndexer: """把切片与向量写入 Chroma 向量库。""" def __init__(self, collection_name: str = "knowledge_base", persist_dir: str = "./storage"): self.client = chromadb.PersistentClient(path=persist_dir) self.collection = self.client.get_or_create_collection(collection_name) def add_chunks(self, chunks: list[Chunk], vectors: list[list[float]]): ids = [c.chunk_id for c in chunks] documents = [c.content for c in chunks] metadatas = [ { "doc_id": c.doc_id, **{k: str(v) for k, v in c.metadata.items()} } for c in chunks ] self.collection.add( ids=ids, documents=documents, embeddings=vectors, metadatas=metadatas ) def query(self, query_vector: list[float], top_k: int = 10): result = self.collection.query( query_embeddings=[query_vector], n_results=top_k ) return result有一个容易踩的坑：Chroma 等向量库的 metadata 字段对数据类型有要求，某些类型会报编码错误，所以我在写入前统一转为字符串。
4.5 检索与重排模块
检索模块不能只依赖向量召回。实际业务中，专业名词、文档编号、精确型号等场景，关键词精确匹配往往比向量召回更可靠。因此，我建议采用多路召回策略：
- 向量召回：处理语义相似但字面不同的查询。
- 关键词召回：处理精确编号、型号、人名等查询。
# 文件路径：app/retrieval/retriever.py class HybridRetriever: """混合检索：向量召回 + 关键词召回，结果合并去重。""" def __init__(self, indexer, embedder, top_k: int = 10): self.indexer = indexer self.embedder = embedder self.top_k = top_k def retrieve(self, query: str) -> list[RetrievalResult]: # 1. 向量召回 query_vector = self.embedder.embed_texts([query])[0] vec_result = self.indexer.query(query_vector, top_k=self.top_k) # 2. 关键词召回 keyword_result = self.indexer.collection.query( query_texts=[query], n_results=self.top_k ) # 3. 合并去重（简化版，生产环境一般按 doc_id + chunk_id 去重） merged = {} for hit in self._extract_hits(vec_result, source="vector"): merged[hit.chunk.chunk_id] = hit for hit in self._extract_hits(keyword_result, source="keyword"): if hit.chunk.chunk_id not in merged: merged[hit.chunk.chunk_id] = hit return list(merged.values())[: self.top_k]重排模块是对多路召回结果做精排的模块。目前主流方案是使用交叉编码器模型，例如 bge-reranker-base。重排的基本思路是：把用户问题和候选文档拼接成一段文本，由模型输出相关性分数，再按分数排序。
# 文件路径：app/retrieval/reranker.py from sentence_transformers import CrossEncoder class Reranker: """基于交叉编码器的重排模块。""" def __init__(self, model_name: str = "BAAI/bge-reranker-base"): self.model = CrossEncoder(model_name, max_length=512) def rerank(self, query: str, documents: list[RetrievalResult], top_k: int = 5) -> list[RetrievalResult]: pairs = [(query, doc.chunk.content) for doc in documents] scores = self.model.predict(pairs) ranked = sorted( zip(documents, scores), key=lambda item: item[1], reverse=True ) return [doc for doc, score in ranked[:top_k]]重排的价值在于：向量召回阶段，模型把每条文本压缩成一个向量，本质是信息有损的。而重排阶段把用户问题和每条候选文档做完整交叉编码，计算量更大，但相关性判断更准确。这是“粗排 + 精排”的经典搜索架构，在 RAG 系统中的价值同样很高。
4.6 生成模块
生成模块负责组装 Prompt 并调用大模型。模块化设计的关键点是 Prompt 模板要独立维护，不能散落在业务代码中。
# 文件路径：app/retrieval/generator.py from openai import OpenAI from app.schema.models import RetrievalResult class Generator: """调用大模型生成答案，并携带引用来源。""" def __init__(self, model_name: str = "gpt-4o-mini", base_url: str = None, api_key: str = None): self.model_name = model_name self.client = OpenAI(base_url=base_url, api_key=api_key) def generate(self, query: str, results: list[RetrievalResult]) -> str: context = "\n\n".join( f"[文档{doc.chunk.metadata.get('doc_id', '未知')}] {doc.chunk.content}" for doc in results ) prompt = f"""你是一个严谨的智能问答助手。请根据给定的知识片段回答用户问题。 要求： 1. 只依据知识片段回答，不要编造片段中没有的内容。 2. 如果片段信息不足，请明确回答“知识库中没有找到相关信息”。 3. 回答末尾标注片段来源。 知识片段： {context} 用户问题： {query} 请给出答案：""" response = self.client.chat.completions.create( model=self.model_name, messages=[{"role": "user", "content": prompt}], temperature=0.2 ) return response.choices[0].message.content生成模块有几个工程细节需要落实：
- temperature 要设置得偏低，保证答案更贴近知识片段，减少发挥。
- Prompt 中明确要求“无法回答时不要编造”，这是控制幻觉的最简单手段。
- 输出引用来源，方便用户核验答案。
如果你希望在答案中返回结构化 JSON，可以把 Prompt 改成要求输出 JSON，再用 JSON 解析器做校验和降级处理。
5. 模块化 RAG 的评估与调优
5.1 为什么 RAG 需要独立评估
很多团队做 RAG 项目时，习惯“感觉回答变好了”或者“看起来挺准”这种主观评价。这是模块化 RAG 项目推进过程中的最大隐患。没有量化指标，你就无法判断项目上线后是变好还是变差，也很难对不同切片策略、不同模型做横向对比。
RAG 评估至少要覆盖三个层面：
- 组件层面：切片器切得是否合理，Embedding 模型语义空间是否准确。
- 检索层面：正确文档有没有被召回，噪声文档占比多少。
- 生成层面：最终答案是否正确、是否忠于知识片段、是否回答了用户问题。
5.2 核心评估指标
RAG 项目中最常用的评估指标可以分为检索指标和生成指标。我整理成一张表：
| 指标名称 | 所属层面 | 含义 | 推荐阈值/目标 |
|---|---|---|---|
| 召回率（Recall@K） | 检索 | 黄金文档是否出现在前 K 个召回结果中 | 越高越好，一般 K=3 时目标 > 0.8 |
| 命中率（Hit Rate） | 检索 | 检索结果中至少包含一条正确文档的比例 | 一般目标 > 0.9 |
| MRR（Mean Reciprocal Rank） | 检索 | 第一条正确文档在结果中的排名倒数均值 | 越高越好，关注是否为 Top1 |
| NDCG@K | 检索 | 检索结果排序质量，兼顾相关性和位置 | 越高越好 |
| 忠实度（Faithfulness） | 生成 | 答案内容是否与知识片段一致，是否存在幻觉 | 一般目标 > 0.85 |
| 答案相关度（Answer Relevance） | 生成 | 答案是否回答了用户的问题 | 一般目标 > 0.9 |
| 上下文相关性（Context Relevance） | 检索→生成 | 提供给模型的上下文与问题的相关程度 | 一般目标 > 0.8 |
5.3 如何理解这些指标并落地调优
明确了指标后，调优就能按数据说话：
- 如果 Hit Rate 偏低，说明向量召回没有带回正确文档，优先检查切片大小、Embedding 模型类型，以及是否需要加入关键词召回。
- 如果 MRR 偏低但 Hit Rate 还可以，说明正确文档存在但排名靠后，应该考虑加大候选数量，并引入重排模型。
- 如果上下文相关性好但答案忠实度低，问题出在 Prompt 模板或者生成模型的参数设置上。
- 如果答案相关度低，但忠实度高，说明模型“忠实”地回答了上下文，上下文却没有覆盖用户真正想要的信息。
实际落地时，可以准备一个包含 100 到 200 条问题的评测集。每条问题标注“正确文档ID”和“标准答案摘要”。每次修改切片参数或更换模型后，都跑一遍评测集，记录指标变化。这个过程自动化后，就是一套简单的 RAG CI 系统。
6. 常见问题与排查思路
模块化 RAG 项目在开发和生产阶段都有一些高频问题。我根据实际经验整理成排查表：
| 问题现象 | 常见原因 | 解决思路 |
|---|---|---|
| 检索不到相关内容 | 切片过大或过小，语义被切散 | 调整 chunk_size 到 300~800，增加重叠 |
| 检索召回一堆无关内容 | 只有向量召回，缺少精确匹配 | 增加关键词召回，引入重排模型 |
| PDF 中文乱码 | 扫描件 PDF，没有文本层 | 先做 OCR，再用文本解析 |
| 答案总是“知识库中没有信息” | 检索失败，或生成模型上下文被截断 | 检查召回 Top1 内容，检查上下文长度 |
| 回答内容与知识片段不一致 | Prompt 指令不明确或温度过高 | 修改 Prompt，降低 temperature |
| 同一问题多次回答结果不稳定 | 生成模型随机性较高 | temperature 尽量降到 0.1~0.2 |
| 新上传文档不生效 | 索引任务未触发或增量逻辑缺失 | 检查离线任务队列，确认文档入库完成 |
| Chroma 写入报错 | metadata 包含非字符串类型 | 写入前统一转为字符串 |
除了表格里的具体问题，我建议每个模块都要在关键节点埋日志，例如“文档解析完成，共解析 20 页”“文本切分完成，生成 120 个切片”“检索完成，返回 10 条结果”。这样即使出了问题，也可以通过日志快速缩小范围。
7. 模块化 RAG 最佳实践与工程建议
7.1 数据质量优先于模型效果
很多 RAG 项目效果不好，根因不是模型不够强，而是知识源本身混乱。建议立项初期先做数据治理，至少要做到：
- 去除重复文档，避免检索结果多个版本互相干扰。
- 明确文档的有效期，过期文档及时下线。
- 对文档做质量分级，核心制度文档优先级高于普通讨论稿。
- 对文档做权限控制，防止越权访问敏感信息。
7.2 离线索引与在线检索分离部署
离线索引涉及大批量 Embedding 计算，耗时长且消耗 CPU/GPU 资源。在线检索则要求低延迟、高并发。两者混部署时，离线批量任务容易拖垮在线服务。建议使用两个服务实例，离线任务通过消息队列触发。
7.3 向量数据库选型要结合规模
项目初期文档量不大，直接用 Chroma 或 FAISS 即可。当文档规模达到百万级，或者需要高并发检索时，建议切换 Milvus、Qdrant 或 Elasticsearch 的向量检索能力。切换时要保证索引器模块的接口不变，只替换实现类。
7.4 关注安全边界
RAG 项目经常处理企业内部文档，直接使用外部大模型服务存在数据合规风险。生产环境建议优先使用私有化部署模型，或者通过云服务商提供的合规通道调用。同时，对用户输入要防止 Prompt 注入，对知识库内容要设计权限过滤。
7.5 增量更新与全量重建
知识库是动态变化的，建议把索引更新分为全量建立和增量更新两类。全量建立适合冷启动和异常恢复，增量更新适合日常同步。增量更新的核心是记录文档变更时间，只处理变化的文件。
7.6 反馈闭环
上线后要记录用户对答案的反馈，例如“有帮助”或“没有帮助”。这些反馈数据是最宝贵的调优信号。反馈可以流入评测集，作为下一轮优化的测试用例。
8. 总结
模块化 RAG 项目展示的并不是某一套代码，而是一整套完整的工程思维。通过把 RAG 流水线拆分成文档解析、文本切分、向量化、索引存储、多路召回、重排、生成和评估这些独立模块，我们才能真正定位问题、优化效果、控制质量。
从这篇文章你可以带走几个核心点：
- 模块化的基础是统一数据结构和清晰接口，而不是依赖某个大而全的框架。
- 文档解析和文本切分是 RAG 质量的基石，值得先投入精力优化。
- 向量召回不是银弹，混合检索加交叉编码器重排才是工程上的常规组合。
- 评估指标是模块化 RAG 项目的“验收单”，没有指标就没有优化闭环。
- 生产环境要考虑数据合规、性能和增量更新，不能只停留在 Demo 阶段。
如果你正准备从零搭建一个 RAG 知识库项目，建议先按文中目录结构搭出骨架，用一批真实业务文档走通全流程，再逐步替换每个模块的实现。RAG 这个领域变化很快，但模块化设计的核心思想不会过时。希望这篇文章能帮你少踩一些坑，把一个能跑的 Demo 打磨成一个能上线的工程。
