模块化RAG项目实战:架构设计、代码实现与评估调优
你好我是老周。在做知识库类项目时经常会出现一个尴尬场景同一个 RAG 系统换了部门文档后效果天差地别甚至同批文档换一种切片策略回答质量就不一样。调参调到最后大家都不知道问题出在文档解析、切片、召回还是重排环节。这篇文章我就围绕“模块化 RAG 项目”这个主题从架构设计、代码实现、评估指标和工程落地四个维度做一个完整拆解。不管你是刚接触 RAG 的新人还是已经在做知识库落地的后端开发都可以照着这套思路去设计和改造自己的项目。1. 什么是模块化 RAG为什么我们需要它1.1 从传统 RAG 到模块化 RAGRAGRetrieval-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可以基于以下依赖构建pymupdf1.23.0 python-docx1.1.0 beautifulsoup44.12.0 langchain-text-splitters0.2.0 chromadb0.4.0 sentence-transformers2.2.0 openai1.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_factorydict) # 来源、作者、时间等元信息 dataclass class Chunk: chunk_id: str # 切片唯一 ID doc_id: str # 所属文档 ID content: str # 切片文本内容 metadata: dict field(default_factorydict) # 页码、章节等定位信息 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(fUnsupported file type: {ext}) def _parse_pdf(self, file_name: str, file_bytes: bytes) - Document: doc fitz.open(streamfile_bytes, filetypepdf) content_parts [] metadata {page_count: len(doc), source: file_name} for page_num, page in enumerate(doc, start1): text page.get_text(text) content_parts.append(f\n 第{page_num}页 \n{text}) content \n.join(content_parts) return Document(doc_idfile_name, titlefile_name, contentcontent, metadatametadata) 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_idfile_name, titlefile_name, contentcontent, metadata{source: file_name}) def _parse_html(self, file_name: str, file_bytes: bytes) - Document: soup BeautifulSoup(file_bytes.decode(utf-8, errorsignore), html.parser) title soup.title.string.strip() if soup.title else file_name content soup.get_text(separator\n, stripTrue) return Document(doc_idfile_name, titletitle, contentcontent, metadata{source: file_name}) def _parse_text(self, file_name: str, file_bytes: bytes) - Document: content file_bytes.decode(utf-8, errorsignore) return Document(doc_idfile_name, titlefile_name, contentcontent, 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_idchunk_id, doc_iddoc_id, contentchunk_text, metadatachunk_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, devicecpu) def embed_texts(self, texts: list[str]) - list[list[float]]: # normalize_embeddingsTrue 会做向量归一化便于内积近似余弦相似度 vectors self.model.encode( texts, normalize_embeddingsTrue, show_progress_barFalse ) 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(pathpersist_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( idsids, documentsdocuments, embeddingsvectors, metadatasmetadatas ) def query(self, query_vector: list[float], top_k: int 10): result self.collection.query( query_embeddings[query_vector], n_resultstop_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_kself.top_k) # 2. 关键词召回 keyword_result self.indexer.collection.query( query_texts[query], n_resultsself.top_k ) # 3. 合并去重简化版生产环境一般按 doc_id chunk_id 去重 merged {} for hit in self._extract_hits(vec_result, sourcevector): merged[hit.chunk.chunk_id] hit for hit in self._extract_hits(keyword_result, sourcekeyword): 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_length512) 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), keylambda item: item[1], reverseTrue ) 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_urlbase_url, api_keyapi_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( modelself.model_name, messages[{role: user, content: prompt}], temperature0.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 项目中最常用的评估指标可以分为检索指标和生成指标。我整理成一张表指标名称所属层面含义推荐阈值/目标召回率RecallK检索黄金文档是否出现在前 K 个召回结果中越高越好一般 K3 时目标 0.8命中率Hit Rate检索检索结果中至少包含一条正确文档的比例一般目标 0.9MRRMean Reciprocal Rank检索第一条正确文档在结果中的排名倒数均值越高越好关注是否为 Top1NDCGK检索检索结果排序质量兼顾相关性和位置越高越好忠实度Faithfulness生成答案内容是否与知识片段一致是否存在幻觉一般目标 0.85答案相关度Answer Relevance生成答案是否回答了用户的问题一般目标 0.9上下文相关性Context Relevance检索→生成提供给模型的上下文与问题的相关程度一般目标 0.85.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 打磨成一个能上线的工程。