在做企业级知识库问答系统时很多RAG项目并不是被大模型能力难住的而是被“项目结构”难住的。Demo阶段把文档加载、文本切分、向量检索、大模型生成全部写在一个文件里确实能跑通一旦进入真实业务数据格式变多、存量文档变大、召回结果不稳定所有问题都堆在同一个函数里排查和改动都非常痛苦。这篇文章围绕“模块化RAG项目”展开讲清楚RAG的核心链路、模块边界和工程化落地方式并使用Python实现一个可替换、可测试的最小RAG工程。无论你是刚接触RAG的初学者还是在做企业内部知识库落地的开发者都可以从这篇文章里找到可以直接参考的项目结构。1. RAG与模块化RAG的核心概念1.1 什么是RAGRAGRetrieval-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_KEYsk-your-key-here OPENAI_BASE_URLhttps://api.openai.com EMBEDDING_MODELtext-embedding-3-small LLM_MODELgpt-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依赖pipelinepipeline依赖corecore内部模块之间尽量不互相耦合。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 检索与重排模块检索模块根据用户问题从向量库召回候选文档。最基础的是向量召回只依赖语义相似度。实际项目中向量召回可能存在两个问题第一关键词完全匹配但语义距离较远的文档可能不被召回第二召回的候选文档数量如果很大噪声会稀释答案。因此常引入混合检索和重排。混合检索是同时执行向量召回和关键词召回再合并结果用RRFReciprocal 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_factorydict) def __repr__(self): return fDocument(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, encodingutf-8) as f: content f.read() return [ Document( contentcontent, 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(contentchunk, metadatametadata)) 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_keyapi_key or os.getenv(OPENAI_API_KEY), base_urlbase_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(modelself.model, inputtexts) 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)), keylambda i: scores[i], reverseTrue, )[: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_kself.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(keylambda x: x[0], reverseTrue) 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_keyapi_key or os.getenv(OPENAI_API_KEY), base_urlbase_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( modelself.model, messages[ {role: system, content: system_prompt}, {role: user, content: user_prompt}, ], temperature0.2, ) return resp.choices[0].message.contentPrompt中使用了[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,