模块化RAG项目实战：构建可替换、可测试的知识库问答系统

在做企业级知识库问答系统时，很多RAG项目并不是被大模型能力难住的，而是被“项目结构”难住的。Demo阶段把文档加载、文本切分、向量检索、大模型生成全部写在一个文件里，确实能跑通；一旦进入真实业务，数据格式变多、存量文档变大、召回结果不稳定，所有问题都堆在同一个函数里，排查和改动都非常痛苦。这篇文章围绕“模块化RAG项目”展开，讲清楚RAG的核心链路、模块边界和工程化落地方式，并使用Python实现一个可替换、可测试的最小RAG工程。无论你是刚接触RAG的初学者，还是在做企业内部知识库落地的开发者，都可以从这篇文章里找到可以直接参考的项目结构。

1. RAG与模块化RAG的核心概念

1.1 什么是RAG

RAG（Retrieval-Augmented Generation，检索增强生成）是一种把“检索”和“生成”结合起来的技术方案。它的基本思路可以这样理解：大模型本身有参数化知识，但训练数据有截止时间，同时也不了解企业内部私有文档。RAG在模型回答之前，先从一个外部知识库中检索出与问题相关的文本片段，再把这些片段作为上下文交给大模型，最终生成答案。

这个流程带来两个明显收益。第一，降低幻觉。模型不再完全依赖“背下来的知识”，而是基于给定的资料回答问题。第二，知识更新成本降低。如果一套制度或说明书发生变化，只需要更新知识库中的文档，不需要重新训练模型。这也是RAG在知识库问答、智能客服、文档解读等场景中被广泛使用的原因。

1.2 什么是模块化RAG

模块化RAG并不是一个新框架，而是一种工程组织方式。它把RAG链路拆成独立的模块，例如文档加载、文本切分、向量化、向量存储、检索、重排、答案生成等，每个模块只负责一个职责，模块之间通过清晰的数据结构进行通信。

为什么要做模块化？最直接的原因是可替换性。开发阶段可以用本地文件加载器，生产阶段要换成S3或数据库数据源；Demo阶段可以用内存向量库，生产阶段要换Milvus或pgvector；今天用OpenAI Embedding，明天可能换成开源BGE模型。如果代码里到处是硬编码，任何替换都会牵一发而动全身。模块化之后，替换实现只需要改一行装配代码，核心逻辑不受影响。

模块化的另一个收益是可测试性。每个模块都可以单独构造输入和预期输出进行测试。比如文本切分模块，可以单独测试边界字符和重叠长度；检索模块可以单独测试top_k结果的准确性。当问答结果变差时，也能快速定位到底哪一层出了问题。

1.3 模块化RAG的典型应用场景

• 企业内部知识库问答：把公司制度、产品文档、FAQ导入知识库，员工用自然语言提问。
• 智能客服：接入工单系统、帮助中心文档，先检索历史解决方案，再由大模型组织回复话术。
• 研发文档问答：让Go、Java、Python等不同技术文档可以被集中检索，辅助代码评审和问题排查。
• 金融合规与银行场景：对风险条款、监管文件、产品说明书做问答和核验，要求答案有出处、可追溯。
• 行业标准与协议问答：例如通信协议标准、技术规范文档的语义检索和对比分析。

在这些场景中，模块化的价值尤其明显。因为不同行业的数据格式差异很大，检索要求也不同，只有把流程解耦，才能针对具体场景做定制优化，同时复用通用的RAG骨架。

2. 环境准备与项目结构

2.1 运行环境说明

本文示例使用Python编写，整体不绑定某个特定RAG框架，而是用最直接的抽象接口演示模块化设计。运行需要准备以下环境：

• Python 3.10 或更高版本。
• pip 包管理器。
• OpenAI SDK，用于调用兼容OpenAI接口的Embedding模型和LLM。
• numpy，用于向量相似度计算。
• 一个可用的Embedding服务和LLM服务，可以是远程API，也可以是本地兼容服务。

版本需要根据实际环境调整。本文示例中默认使用OpenAI兼容SDK，如果你使用的是本地模型或企业内部模型网关，只需要把base_url指向对应地址即可，不改变整体模块结构。建议在项目根目录创建虚拟环境，避免依赖冲突。

创建虚拟环境和安装依赖的命令如下：

python3 -m venv venv source venv/bin/activate pip install openai numpy python-dotenv

为了便于管理密钥，建议准备一个.env文件，把API Key和模型名放到环境变量里。下面是一个.env.example示例：

OPENAI_API_KEY=sk-your-key-here OPENAI_BASE_URL=https://api.openai.com EMBEDDING_MODEL=text-embedding-3-small LLM_MODEL=gpt-4o-mini

如果你的模型服务支持OpenAI协议但地址不同，只需要修改OPENAI_BASE_URL。需要注意，API Key属于敏感信息，任何情况下都不应该提交到Git仓库中，.env文件要加入.gitignore。

2.2 项目目录结构

一个清晰的目录结构是模块化项目的基础。本文项目结构如下：

modular_rag/ ├── config/ │ └── settings.py # 全局配置 ├── core/ │ ├── documents.py # Document数据模型 │ ├── loader/ │ │ └── text_loader.py # 文本文件加载器 │ ├── splitter/ │ │ └── text_splitter.py # 文本切分器 │ ├── embedding/ │ │ └── embedding_client.py # Embedding客户端 │ ├── vector_store/ │ │ ├── base_store.py # 向量库抽象接口 │ │ └── memory_store.py # 内存向量库实现 │ ├── retriever/ │ │ └── retriever.py # 检索器 │ ├── reranker/ │ │ └── reranker.py # 重排器 │ └── generator/ │ └── llm_generator.py # 答案生成器 ├── pipeline/ │ └── rag_pipeline.py # RAG流水线组装 ├── app/ │ ├── ingest.py # 文档入库入口 │ └── query.py # 问答入口 └── data/ └── demo.md # 示例文档

这个结构把模块分为三层：

• core层：核心能力组件，每个目录对应一个独立模块。
• pipeline层：把核心组件串成一条完整RAG流程。
• app层：面向使用者的命令行入口。

这样的分层让代码的依赖方向变得单向：app依赖pipeline，pipeline依赖core，core内部模块之间尽量不互相耦合。

3. 核心模块设计与原理拆解

3.1 文档加载模块

文档加载模块的职责是把不同来源的数据统一成一种内部数据结构。来源可能是本地文本文件、PDF、Word、网页、数据库查询结果，也可能是对象存储中的文件。如果每种来源都返回不同格式，后续模块就需要大量if-else判断，代码会很难维护。

解决方案是定义一个Document数据结构，至少包含两个字段：

• content：文本内容，这是后续切分和向量化的主要对象。
• metadata：元数据，例如来源路径、页码、标题、创建时间等。元数据会在最终回答中用于溯源展示。

加载模块只需要暴露一个统一方法，输入是“来源标识”，输出是List[Document]。PDF等复杂格式的解析可以放在具体实现中，但对外接口保持一致。这样业务代码不关心文档来自哪里，只关心拿到了Document列表。

3.2 文本切分模块

文本切分是RAG里最容易被低估的模块。切得太粗，一个chunk可能包含多个主题，语义不聚焦，检索时匹配不精准；切得太细，上下文被打断，可能丢失关键信息，同时产生大量碎片，增加存储和检索成本。

常用的切分策略有两种。第一种是固定长度切分，使用chunk_size控制每个文本块的最大字符数，使用chunk_overlap控制相邻块之间的重叠长度，保证跨边界的关键信息不被切断。第二种是按结构切分，例如Markdown标题、PDF章节、代码函数等，按结构化边界切分，检索结果更符合原文逻辑。实践中经常把两种策略结合：先按标题或段落做粗切分，再对超长段落做固定长度细切分。

chunk_size的选择没有绝对标准，需要结合Embedding模型的最大输入长度和业务数据特点调整。中文场景下，一个500字符的chunk大约能覆盖一段完整描述。chunk_overlap一般设置为chunk_size的10%到20%，太小起不到衔接作用，太大会导致重复内容过多。

3.3 向量化模块

向量化模块把文本转换成高维向量，让语义相近的文本在向量空间中距离更近。这是语义检索的基础。常见的Embedding模型包括OpenAI的text-embedding-3-small、开源社区常用的BGE系列、M3E、bge-m3等。选择模型时主要看三件事：语义效果、向量维度、输入长度限制。

向量化模块需要提供两个方法：一个是embed_documents，批量把chunk列表转成向量列表；另一个是embed_query，把查询问题转成向量。之所以分成两个方法，是因为不少Embedding模型对“长文本”和“短查询”有不同的编码策略，分开封装便于后续调整。

批量编码时可以一次传入多段文本，减少网络请求次数，提高吞吐量。对于异步场景，还需要考虑并发控制和失败重试，这部分放到工程化实践中说明。

3.4 向量存储模块

向量存储模块负责保存文本chunk和对应的向量，并提供相似度检索能力。开发阶段可以使用内存向量库，把所有向量存在Python列表里；生产阶段需要支持持久化、大规模数据和高并发检索的向量数据库，例如Milvus、Qdrant、pgvector、Elasticsearch向量索引等。

向量数据库内部通常使用HNSW、IVF等近似最近邻索引来加速检索。调用方一般不需要关心索引细节，但需要知道相似度计算方式。常见的相似度算法有余弦相似度、内积、欧氏距离。语义检索场景最常用的是余弦相似度，它只关注向量的方向，不受向量长度影响。

向量存储模块需要向外部隐藏具体数据库差异。上层只调用add_documents和search两个方法，具体是写内存列表还是写Milvus集合，完全由实现类决定。

3.5 检索与重排模块

检索模块根据用户问题从向量库召回候选文档。最基础的是向量召回，只依赖语义相似度。实际项目中，向量召回可能存在两个问题：第一，关键词完全匹配但语义距离较远的文档可能不被召回；第二，召回的候选文档数量如果很大，噪声会稀释答案。因此常引入混合检索和重排。

混合检索是同时执行向量召回和关键词召回，再合并结果，用RRF（Reciprocal Rank Fusion）等算法融合排序。关键词召回可以使用BM25，向量召回使用Embedding相似度，两者互补。重排模块则是对第一轮召回的候选文档做更精细的排序，通常使用cross-encoder模型，把“问题-文档”拼接后一起编码，交互式计算相关性分数，比bi-encoder的二阶段向量比对更准确。

检索模块的接口是输入查询文本，输出候选Document列表。重排模块的接口是输入查询和候选列表，输出重新排序后的列表。这样可以把检索策略和排序策略完全解耦。

3.6 生成模块

生成模块把检索到的文本片段和用户问题组装成Prompt，交给大模型生成答案。Prompt设计需要明确告诉模型：只根据给定资料回答，资料中没有的信息不要编造，必要时明确回答“未找到相关信息”。

同时，Prompt中要给每个chunk编号，方便模型在回答时引用出处。生成模块还可以接入后续校验逻辑，例如用规则判断答案是否包含资料之外的实体，从而降低幻觉。这个模块是RAG链路面向用户的最后一步，也是直接决定体验的环节。

4. 完整实战案例：构建一个可插拔的模块化RAG项目

4.1 创建项目结构

首先在本地创建目录结构。在命令行依次执行：

mkdir -p modular_rag/config mkdir -p modular_rag/core/loader mkdir -p modular_rag/core/splitter mkdir -p modular_rag/core/embedding mkdir -p modular_rag/core/vector_store mkdir -p modular_rag/core/retriever mkdir -p modular_rag/core/reranker mkdir -p modular_rag/core/generator mkdir -p modular_rag/pipeline mkdir -p modular_rag/app mkdir -p modular_rag/data

然后按后续代码逐步创建文件。每个文件的路径会在代码注释中标注。

4.2 全局配置模块

先创建config/settings.py，统一管理模型名、切分参数、检索参数等配置。配置全部从环境变量读取，并提供默认值。

# 文件路径：modular_rag/config/settings.py """全局配置管理模块。""" import os from dotenv import load_dotenv # 自动加载项目根目录下的 .env 文件 load_dotenv() class Settings: # Embedding 与 LLM 配置 EMBEDDING_MODEL = os.getenv("EMBEDDING_MODEL", "text-embedding-3-small") LLM_MODEL = os.getenv("LLM_MODEL", "gpt-4o-mini") # 文本切分参数 CHUNK_SIZE = int(os.getenv("CHUNK_SIZE", "500")) CHUNK_OVERLAP = int(os.getenv("CHUNK_OVERLAP", "50")) # 检索与重排参数 RETRIEVE_TOP_K = int(os.getenv("RETRIEVE_TOP_K", "5")) RERANK_TOP_K = int(os.getenv("RERANK_TOP_K", "3"))

这段代码的关键点在于：所有参数都放在一个配置类中，后续替换模型或调整切分大小，不需要修改业务代码。load_dotenv()会自动读取项目根目录的.env文件，降低本地开发的配置成本。

4.3 文档数据模型与加载模块

core/documents.py定义统一的文档数据模型。

# 文件路径：modular_rag/core/documents.py """文档数据模型，所有数据源统一转换成该结构。""" from dataclasses import dataclass, field from typing import Dict @dataclass class Document: content: str metadata: Dict[str, str] = field(default_factory=dict) def __repr__(self): return f"Document(metadata={self.metadata}, content={self.content[:50]}...)"

然后创建core/loader/text_loader.py，实现一个最简单的本地文本文件加载器。为了让模块结构更清晰，先定义一个抽象基类BaseLoader。

# 文件路径：modular_rag/core/loader/text_loader.py """文档加载模块：从不同来源加载文档，返回Document列表。""" from abc import ABC, abstractmethod from typing import List from core.documents import Document class BaseLoader(ABC): """文档加载器抽象接口。""" @abstractmethod def load(self, source: str) -> List[Document]: """加载指定来源的文档，返回Document列表。""" raise NotImplementedError class TextFileLoader(BaseLoader): """加载本地文本文件。""" def load(self, source: str) -> List[Document]: with open(source, "r", encoding="utf-8") as f: content = f.read() return [ Document( content=content, metadata={"source": source}, ) ]

这个模块体现了抽象接口的价值：如果后续需要加载PDF，只需要新增一个PdfLoader，实现同一个load方法，上层代码不需要感知差异。

4.4 文本切分模块

core/splitter/text_splitter.py实现基于字符长度的文本切分。

# 文件路径：modular_rag/core/splitter/text_splitter.py """文本切分模块：将长文档切成适合嵌入和检索的chunk。""" from typing import List from core.documents import Document class TextSplitter: def __init__(self, chunk_size: int = 500, chunk_overlap: int = 50): self.chunk_size = chunk_size self.chunk_overlap = chunk_overlap def split_text(self, text: str) -> List[str]: """将单段文本按固定长度切分。""" if not text: return [] chunks = [] start = 0 while start < len(text): end = start + self.chunk_size chunks.append(text[start:end]) if end >= len(text): break start = end - self.chunk_overlap return chunks def split_documents(self, documents: List[Document]) -> List[Document]: """将多个文档切分成带元数据的chunk。""" result = [] for doc in documents: for chunk in self.split_text(doc.content): metadata = dict(doc.metadata) metadata["chunk_index"] = len(result) result.append(Document(content=chunk, metadata=metadata)) return result

这里的实现是教学用的简化版。生产环境建议在切分前先按段落、标题等结构边界做预处理，避免把完全无关的内容拼在同一个chunk中。chunk_index用于记录顺序，方便后续溯源和调试。

4.5 向量化模块

core/embedding/embedding_client.py封装OpenAI兼容的Embedding接口。

# 文件路径：modular_rag/core/embedding/embedding_client.py """向量化模块：将文本转换为语义向量。""" import os from typing import List from openai import OpenAI class OpenAIEmbeddingClient: def __init__( self, model: str = "text-embedding-3-small", api_key: str = None, base_url: str = None, ): self.client = OpenAI( api_key=api_key or os.getenv("OPENAI_API_KEY"), base_url=base_url or os.getenv("OPENAI_BASE_URL"), ) self.model = model def embed_documents(self, texts: List[str]) -> List[List[float]]: """批量将文本列表转换为向量。""" if not texts: return [] resp = self.client.embeddings.create(model=self.model, input=texts) return [item.embedding for item in resp.data] def embed_query(self, text: str) -> List[float]: """将查询文本转换为向量。""" return self.embed_documents([text])[0]

embed_documents和embed_query分开设计，是因为在某些Embedding服务中查询向量与文档向量可能使用不同的编码模式。api_key和base_url支持从环境变量读取，也可以显式传入，方便测试时替换Mock对象。

4.6 向量存储模块

先定义抽象接口core/vector_store/base_store.py。

# 文件路径：modular_rag/core/vector_store/base_store.py """向量存储抽象接口。""" from abc import ABC, abstractmethod from typing import List from core.documents import Document class BaseVectorStore(ABC): @abstractmethod def add_documents( self, documents: List[Document], embeddings: List[List[float]], ) -> None: """将文档和对应向量写入向量库。""" raise NotImplementedError @abstractmethod def search( self, query_embedding: List[float], top_k: int = 5, ) -> List[Document]: """根据查询向量召回最相似的文档。""" raise NotImplementedError

再实现内存版core/vector_store/memory_store.py。这个实现不需要外部数据库，方便学习和验证。

# 文件路径：modular_rag/core/vector_store/memory_store.py """内存向量库实现，适合开发调试和小规模演示。""" from typing import List import numpy as np from core.documents import Document from core.vector_store.base_store import BaseVectorStore class InMemoryVectorStore(BaseVectorStore): def __init__(self): self.documents: List[Document] = [] self.embeddings: List[List[float]] = [] def add_documents( self, documents: List[Document], embeddings: List[List[float]], ) -> None: self.documents.extend(documents) self.embeddings.extend(embeddings) def search( self, query_embedding: List[float], top_k: int = 5, ) -> List[Document]: if not self.embeddings: return [] scores = [ self._cosine_similarity(query_embedding, emb) for emb in self.embeddings ] top_indices = sorted( range(len(scores)), key=lambda i: scores[i], reverse=True, )[:top_k] return [self.documents[i] for i in top_indices] @staticmethod def _cosine_similarity( vec_a: List[float], vec_b: List[float], ) -> float: a = np.asarray(vec_a) b = np.asarray(vec_b) return float( np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b) + 1e-10) )

这里手动实现了余弦相似度，并对分母做了平滑处理。生产环境中使用Milvus等数据库时，相似度计算由数据库内部完成，不需要自己实现。

4.7 检索与重排模块

检索器core/retriever/retriever.py负责调用向量库查询。

# 文件路径：modular_rag/core/retriever/retriever.py """检索模块：根据查询问题召回候选文档。""" from typing import List from core.documents import Document from core.embedding.embedding_client import OpenAIEmbeddingClient from core.vector_store.base_store import BaseVectorStore class VectorRetriever: def __init__( self, vector_store: BaseVectorStore, embedding_client: OpenAIEmbeddingClient, top_k: int = 5, ): self.vector_store = vector_store self.embedding_client = embedding_client self.top_k = top_k def retrieve(self, query: str) -> List[Document]: query_embedding = self.embedding_client.embed_query(query) return self.vector_store.search(query_embedding, top_k=self.top_k)

重排器core/reranker/reranker.py先实现一个基于关键词重合度的轻量版，用于说明重排流程。真正生产环境建议换成cross-encoder模型。

# 文件路径：modular_rag/core/reranker/reranker.py """重排模块：对第一轮候选文档进行精细化排序。""" from abc import ABC, abstractmethod from typing import List from core.documents import Document class BaseReranker(ABC): @abstractmethod def rerank( self, query: str, documents: List[Document], top_k: int = 3, ) -> List[Document]: raise NotImplementedError class KeywordReranker(BaseReranker): """ 一个简单的教学版重排器。 生产环境建议替换为 BGE-Reranker 等 cross-encoder 模型。 """ def rerank( self, query: str, documents: List[Document], top_k: int = 3, ) -> List[Document]: query_terms = set(query.lower().split()) scored = [] for doc in documents: content_lower = doc.content.lower() hit_count = sum(1 for term in query_terms if term in content_lower) scored.append((hit_count, doc)) scored.sort(key=lambda x: x[0], reverse=True) return [doc for _, doc in scored[:top_k]]

这个重排器虽然简单，但已经演示了“排序器”的接口定位。后续替换成真正的模型重排时，rerank方法签名不变，上层不需要修改。

4.8 生成模块

core/generator/llm_generator.py负责将上下文和问题组装成Prompt，并调用LLM生成答案。

# 文件路径：modular_rag/core/generator/llm_generator.py """答案生成模块：基于检索上下文生成最终回答。""" import os from typing import List from openai import OpenAI from core.documents import Document class LLMGenerator: def __init__( self, model: str = "gpt-4o-mini", api_key: str = None, base_url: str = None, ): self.client = OpenAI( api_key=api_key or os.getenv("OPENAI_API_KEY"), base_url=base_url or os.getenv("OPENAI_BASE_URL"), ) self.model = model def generate( self, query: str, contexts: List[Document], ) -> str: context_text = "\n\n".join( [f"[{i + 1}] {doc.content}" for i, doc in enumerate(contexts)] ) system_prompt = ( "你是一个严谨的知识库问答助手。请只根据提供的资料回答问题；" "如果资料中没有相关信息，请直接回答“未找到相关信息”，" "不要编造内容，不要输出资料之外的推测。" ) user_prompt = f"问题：{query}\n\n可用资料：\n{context_text}" resp = self.client.chat.completions.create( model=self.model, messages=[ {"role": "system", "content": system_prompt}, {"role": "user", "content": user_prompt}, ], temperature=0.2, ) return resp.choices[0].message.content

Prompt中使用了[1] [2]这样的编号，目的是让模型在回答时可以引用来源编号。实际产品中，可以在回答后把编号对应的文档来源展示给用户，增强可追溯性。

4.9 RAG流水线与入口

pipeline/rag_pipeline.py将上述模块串起来，对外暴露两个方法：ingest用于文档入库，query用于问答。

# 文件路径：modular_rag/pipeline/rag_pipeline.py """RAG流水线组装模块。""" from typing import List from core.documents import Document from core.embedding.embedding_client import OpenAIEmbeddingClient from core.generator.llm_generator import LLMGenerator from core.loader.text_loader import BaseLoader from core.reranker.reranker import BaseReranker from core.retriever.retriever import VectorRetriever from core.splitter.text_splitter import TextSplitter from core.vector_store.base_store import BaseVectorStore class RAGPipeline: def __init__( self, loader: BaseLoader, splitter: TextSplitter, embedding_client: OpenAIEmbeddingClient, vector_store: BaseVectorStore, retriever: VectorRetriever, reranker: BaseReranker, generator: LLMGenerator, ): self.loader = loader self.splitter = splitter self.embedding_client = embedding_client self.vector_store = vector_store self.retriever = retriever self.reranker = reranker self.generator = generator def ingest(self, source: str) -> int: """加载文档并写入向量库，返回写入的chunk数量。""" documents = self.loader.load(source) chunks = self.splitter.split_documents(documents) texts = [chunk.content for chunk in chunks] embeddings = self.embedding_client.embed_documents(texts) self.vector_store.add_documents(chunks, embeddings) return len(chunks) def query(self,