ARTICLE · INTELLIGENCE

战地情报 · 详情页

来自尧图项目组的一线实战观察与深度解析

RAG系统从0到生产:原理拆解、代码实现与工程落地全攻略

RAG系统从0到生产:原理拆解、代码实现与工程落地全攻略

做RAG系统做到吐之后,我决定把从0到生产这条路上的所有细节一次讲清楚。不管你是刚接触大模型应用开发的新手,还是已经在做RAG但是效果一直不理想的工程师,这篇文章围绕检索增强生成(Retrieval-Augmented Generation)的完整流程,从原理拆解、代码实现到工程落地,把那些文档里不讲、实战里才有的东西全翻出来。

先说结论:RAG的本质不是让大模型“变聪明”,而是给它配一个随取随用的外部知识库,让它在回答之前先去查资料,再基于查到的内容作答。这个思路看起来简单,但真正落地的时候,分块怎么切、向量怎么存、检索怎么排序、上下文怎么拼,每一步都会直接影响最终效果。我带着你从零开始把整条链路走一遍,代码直接给你,工程实践中的坑也帮你标好。

1. 先搞清楚:RAG到底解决什么问题

1.1 大模型天生的两个短板

大模型虽然知道很多东西,但有两个绕不开的死穴。第一是知识时效性,绝大多数通用模型的训练数据截止在一个固定时间点,之后发生的事情它一概不知。你问它最新的活动安排、最新的产品参数、最新发布的政策文件,它要么说不知道,要么凭印象编一个看起来合理但其实错误的内容。第二是领域知识缺失,通用的预训练模型对特定企业内部的知识——比如老员工脑中的项目经验、散布在各处的非结构化文档、产品线的历史决策记录——基本一窍不通。你让它回答一个只存在于内部资料里的问题,它只能从自己的参数里去猜,猜不出来就一本正经地胡编。

这种胡编现象在行业里有个专门的说法叫“幻觉”,是所有大模型落地应用的第一大障碍。早期的解决办法是微调,把领域知识通过额外训练的方式塞进模型的参数里。但微调的问题也很明显:每次知识更新都要重新训练,成本高、周期长,而且知识是以参数的形式模糊存储的,模型记没记住、记准没记准,你根本没法保证。

1.2 RAG的解题思路

RAG的思路完全不同:既然模型不知道,那就在它回答之前先告诉它。整个流程可以简化成“查资料然后作答”两步。用户提出一个问题,系统先去一个外部知识库(通常是向量数据库)里检索出最相关的若干段文本,然后把这些文本作为附加上下文,连同用户的问题一起交给大模型,让模型“只依据提供的资料”来组织答案。

这里最核心的转变是,知识不再存储在大模型内部,而是存储在外部。每次回答之前,系统动态地把最需要的知识取出来、拼进提示词里。知识更新就变成了一件低成本的事情:文档入库的时候做了什么修改,检索结果就是什么,不需要重新训练模型。这也是为什么RAG在过去一段时间里成为企业落地大模型应用的主流方案——成本可控、效果可预期、迭代足够快。

1.3 什么样的场景适合RAG

不是所有场景都需要RAG。如果你的应用完全不依赖外部资料,比如让模型做一个通用的文案润色、代码注释生成,那么纯Prompt调用就够了,引入RAG只会增加延迟和复杂度。但当你有这几个需求中的一个,RAG几乎就是刚需:回答必须附上知识来源、知识内容需要频繁更新、问题答案只存在于私有文档中、对回答准确性要求高。典型的例子包括企业内部的智能问答助手、法律和金融文档的条款解析、产品手册的自动客服、学术论文的检索问答,这些都是RAG的主场。

另外要区分的是,RAG不等于“聊天机器人加了一个数据库”,它是一整套检索与生成协同的工程架构。后续的问题在于:文档怎么准备、检索质量怎么保证、生成结果怎么约束。下面我先把整条流程拆开,然后直接上代码。

2. 从0到1:RAG全链路代码实现

2.1 环境准备与依赖安装

我用Python实现,选型的核心工具尽量保证轻量通用:文档处理用LangChain作为编排框架,向量存储用Chroma(本地开发最省事),向量模型用开源的Embedding模型,生成部分接大模型API。你可以根据自己的实际情况替换任何一个组件,因为RAG的每一环基本都是可插拔的。

pip install langchain langchain-community langchain-openai chromadb pip install pypdf python-docx # 文档解析用 pip install sentence-transformers # 本地向量模型加载

我这里用的是OpenAI兼容的API来调大模型,后面你换成任意一家国产模型、本地部署的模型,只要接口兼容,代码基本不用动。Embedding模型我建议本地运行,原因后面讲到成本的时候会展开。

2.2 文档加载与分块

RAG输入的第一步是把原始文档读进来。这里有个反直觉的事情:很多人以为把PDF、Word里的内容直接塞给模型就行,实则不行。原因在于大模型的上下文窗口有长度限制,一份合同几十页,全塞进去既浪费token,检索时也完全没法命中关键段落。处理方式是先加载、再切块。

先看加载的代码,以最常见的PDF和Markdown为例:

from langchain_community.document_loaders import PyPDFLoader, TextLoader, UnstructuredMarkdownLoader # 加载PDF loader = PyPDFLoader("./docs/产品手册.pdf") docs = loader.load() # 加载Markdown md_loader = UnstructuredMarkdownLoader("./docs/项目说明.md") md_docs = md_loader.load() # 加载纯文本 txt_loader = TextLoader("./docs/会议纪要.txt") txt_docs = txt_loader.load()

加载出来的每个Document对象包含两个重要属性:page_content(文本内容)和metadata(来源、页码等元信息)。元信息很重要,后面做引用溯源全靠它。

接下来是分块。分块是整个RAG流程里第一个决定成败的关键环节。如果块太大,一个块里包含多个主题,检索时会返回大量无关内容;如果块太小,可能单个块连一个完整的事实都装不下。我的经验是从一个基准值开始,再根据文档类型调整:

from langchain.text_splitter import RecursiveCharacterTextSplitter text_splitter = RecursiveCharacterTextSplitter( chunk_size=800, # 每个块的最大字符数 chunk_overlap=150, # 相邻块之间的重叠字符数 length_function=len, separators=["\n\n", "\n", "。", "!", "?", ";", " ", ""] ) chunks = text_splitter.split_documents(docs) print(f"切分为 {len(chunks)} 个文本块")

RecursiveCharacterTextSplitter的工作方式是:优先尝试用最长的分隔符切分,发现切出来的块还是太长,就换更短的分隔符继续切。这个“递归”逻辑保证了文本在尽量接近语义边界的位置断开。chunk_overlap之所以要设置成非零,是为了避免一句话刚开头在一个块里、结尾在下一个块里,检索时哪半边都不完整,导致信息断裂。

这里给你一个实际案例的参考:我处理一份技术架构文档时,起初用chunk_size=500,结果检索精度很不稳定,很多本应命中的内容因为上下文被切断而排名靠后。调到800并且把标点符号加入分隔符后,情况明显好转。不同场景的推荐值后面专门整理一个表。

2.3 向量化与向量库存入

分完块之后,需要把每个文本块转成向量。这里的目的是把文字变成一串数字,让语义相近的文本在数字空间的距离也更近。这个环节的选型直接影响检索质量,我的建议是不要用大模型的Embedding接口来做批量入库,原因只有一个——成本。

本地部署一个开源Embedding模型,单次推理成本几乎为零,而且中文效果已经够用。这里用text2vec-base-chinese作为示例,也可以换成BGE系列:

from langchain_community.embeddings import HuggingFaceEmbeddings embeddings = HuggingFaceEmbeddings( model_name="GanymedeNil/text2vec-large-chinese", model_kwargs={"device": "cpu"}, encode_kwargs={"normalize_embeddings": False} ) # 查看向量维度 sample_vector = embeddings.embed_query("测试一下向量维度") print(f"向量维度: {len(sample_vector)}")

然后创建向量库并写入数据。Chroma在本地开发阶段非常好用,不需要单独起服务,直接嵌到进程里:

from langchain_community.vectorstores import Chroma vectorstore = Chroma.from_documents( documents=chunks, embedding=embeddings, persist_directory="./chroma_db", collection_name="knowledge_base" ) print(f"已入库 {vectorstore._collection.count()} 条文本块")

入库之后需要做一次持久化验证,确认下次能从磁盘直接加载而不是重新向量化:

vectorstore = Chroma( embedding_function=embeddings, persist_directory="./chroma_db", collection_name="knowledge_base" ) print(f"加载成功,库内文本块数: {vectorstore._collection.count()}")

这里有一个容易忽略的细节:Embedding模型的输出向量是有维度的,这个维度由模型决定,和文本长度无关。换模型的时候,向量库里的旧数据和新的向量维度不一致,会直接报错或者检索失效。所以换Embedding模型通常意味着需要重建整个向量库,在生产环境这是个不小的工程,选型时要慎重。

2.4 检索与生成:完整问答链路

入库完成之后就到了最关键的部分——问答时怎么取、怎么用。先做一个简单的相似度检索:

question = "产品的保修政策是什么?" retrieved_docs = vectorstore.similarity_search(question, k=4) for i, doc in enumerate(retrieved_docs): print(f"--- 第{i+1}条 ---") print(doc.page_content) print(f"来源: {doc.metadata}")

similarity_search默认使用余弦相似度或者内积距离,取决于向量库的配置。返回的结果按距离升序排列。但直接拿相似度检索的结果去生成,效果通常不够好,原因后面讲工程化时会具体分析。

生成部分是把检索结果和用户问题拼接成提示词,交给大模型。这里如果用LangChain,代码可以更优雅地串联起来:

from langchain_openai import ChatOpenAI from langchain.prompts import PromptTemplate from langchain_core.runnables import RunnableLambda, RunnablePassthrough from langchain_core.output_parsers import StrOutputParser llm = ChatOpenAI( model="gpt-4o-mini", temperature=0.1, api_key=os.environ["OPENAI_API_KEY"], base_url=os.environ.get("OPENAI_BASE_URL") ) prompt = PromptTemplate.from_template( """你是一名严谨的客服助手。请根据以下资料回答问题,不要编造资料中不存在的信息。 如果资料不足以回答问题,请明确说明“资料中未找到相关信息”。 参考资料: {context} 问题:{question} 回答:""" ) def format_docs(docs): return "\n\n".join(f"[来源{d.metadata.get('source', '未知')}]\n{d.page_content}" for d in docs) rag_chain = ( { "context": vectorstore.as_retriever(k=4) | format_docs, "question": RunnablePassthrough() } | prompt | llm | StrOutputParser() ) result = rag_chain.invoke("产品的保修政策是什么?") print(result)

这个链路的关键是vectorstore.as_retriever(k=4)先执行检索,检出来的文档经过format_docs拼成带格式的上下文,和问题一起填进PromptTemplate,最后让大模型生成回答。把temperature调到0.1以下是为了降低随机性,保证回答尽量忠实于资料。

这里我必须强调一个看似微小但实际非常重要的细节:提示词里的“不要编造资料中不存在的信息”这句话,不是加了一句就完事。模型对这句话的服从程度与参考资料的条理性直接相关。如果你的检索结果里混入了大量不相关的内容,模型很容易被带偏。所以后期你可能会发现,问题主要不是出在生成上,而是出在检索上。

3. 工程化落地:从能跑通到能上线

3.1 分块策略:效果的第一个分水岭

代码跑通之后,你会很快发现一个残酷的事实:demo能做出来,不代表效果能上线。实际测试过程中最大的问题往往就出在分块策略上。很多人用固定chunk_size=500一把梭,结果检索出来的内容要么语义不完整,要么主题混着多个话题。

分块策略没有绝对正确的答案,但有相对合理的出发点。我的建议是:先看文档类型。手册、合同、论文这类本身就有明显章节结构的文档,用MarkdownHeaderTextSplitter或者按标题层级来切,效果远好于纯按字符数硬切。对话记录、会议纪要这类没有明确结构的口语化内容,适合按语义窗口来切,配合合理的重叠。

实际测试中,我踩过一个典型的坑:用固定800字符切一份产品使用手册,其中有一段是把“保修政策”和“联系客服”写在相邻的块里,结果用户问“如何联系客服”时,检索出来的内容只有保修政策,模型拿着这半截内容作答,答得又偏又误导人。后来改成按标题层级切,这个问题直接消失。

一个相对稳妥的分块参数组合我帮你整理成表,可以参考:

文档类型建议分块方法chunk_sizechunk_overlap
技术文档/白皮书按Markdown标题层级切800-1200150-200
合同/法律条款按段落+标点切600-1000100-150
会议纪要/对话记录按说话人轮次切400-60050-100
FAQ/问答对一个问答对作为一个整体200-4000
新闻资讯按段落切600-800100

记住一个原则:让每个文本块尽量成为一个“自洽的语义单元”,也就是单独拿出去也能被人看懂、能回答一个完整问题的程度。这比任何参数都重要。

3.2 混合检索与重排序:生产级的检索优化

如果只做向量检索,你会遇到一个经典困境:两个问题表面上措辞不同,但意思相同,向量检索能找到;两个问题措辞高度相似但意思完全不同,甚至问题里包含的特定术语只在一段话里出现,向量检索往往抓不准。

单一的向量检索在这类场景下是有局限的。行业里通用的解法是“混合检索”:先同时用向量检索和关键词检索(BM25)各取一批结果,再把两批结果合并去重,最后用重排序模型(Reranker)做精排。

先看BM25的引入方式:

# 如果没有独立的BM25服务,本地可以先用rank_bm25实现 pip install rank-bm25 from rank_bm25 import BM25Okapi import jieba # 假设tokenized_docs是分词后的文档列表 bm25 = BM25Okapi(tokenized_docs) tokenized_query = list(jieba.cut(question)) scores = bm25.get_scores(tokenized_query)

但这里我不想把rank_bm25写成主力方案,因为它只适合一次性处理小规模文档。生产环境里,Elasticsearch内置的BM25实现才是主流选择。向量检索和关键词检索各跑一遍,拿到两类候选结果,接下来用一个Reranker模型对融合后的候选集重新打分排序。

Reranker是近一两年RAG工程里非常关键的优化点。它和Embedding模型的区别在于:Embedding是预先计算好所有文本的向量,检索时只计算向量距离;Reranker接收“问题+文档片段”作为输入,直接输出相关性分数。本质上是让模型对每一对问题进行深度语义匹配,效果更好,但代价是慢,所以只对少量候选做精排。

我用bge-reranker-large做了一个简单的精排层:

from transformers import AutoModelForSequenceClassification, AutoTokenizer reranker_tokenizer = AutoTokenizer.from_pretrained("BAAI/bge-reranker-large") reranker_model = AutoModelForSequenceClassification.from_pretrained("BAAI/bge-reranker-large") def rerank(query, documents, top_k=4): pairs = [[query, doc.page_content] for doc in documents] inputs = reranker_tokenizer(pairs, padding=True, truncation=True, return_tensors="pt", max_length=512) scores = reranker_model(**inputs).logits.view(-1).detach().numpy() scored_docs = sorted(zip(documents, scores), key=lambda x: x[1], reverse=True) return [doc for doc, score in scored_docs[:top_k]]

配合主检索流程:

from langchain_community.retrievers import BM25Retriever vector_retriever = vectorstore.as_retriever(search_type="similarity", search_kwargs={"k": 10}) bm25_retriever = BM25Retriever.from_documents(chunks) bm25_retriever.k = 10 vector_docs = vector_retriever.invoke(question) bm25_docs = bm25_retriever.invoke(question) # 合并去重 merged = {doc.metadata.get("source", "") + doc.page_content[:100]: doc for doc in vector_docs + bm25_docs} unique_docs = list(merged.values()) # 重排序取前4 final_docs = rerank(question, unique_docs, top_k=4)

引入Reranker之后的效果提升是很明显的。我用同一组测试集做过对比:只做向量检索的召回命中率约在70%,加上BM25召回后提升了5个百分点,再加Reranker精排后直接跳到90%以上。这个优化链条值得花时间做。

3.3 评估指标:用数据说话才能迭代

RAG项目的另一个常见问题是“凭感觉调参”。今天觉得回答好像好了,明天换个问题又崩了。没有量化评估,你永远不知道改进方向在哪。

RAG的评估主要分两段:检索段和生成段。检索段看两个指标:召回率(正确答案是否存在于检索出来的结果中)和准确率(检索结果中有几条是真正相关的)。生成段看最终答案和标准答案之间的语义相似度、忠实度(是否与参考资料一致)、答案完整性。

工程上如果要做正式评估,可以构建一份测试集:至少准备100个问答对,每个问题标注出处文档。然后用这批数据跑整体流程,计算指标,看看调整哪个环节能带来数据上的提升。举例:如果你发现recall很高但答案质量差,那问题多半出在生成阶段的提示词或者上下文组织;如果recall本身就低,那应该优先优化分块和检索。

自建评估集虽然费时间,但它能帮你打破“感觉良好”的恶性循环,这也是一个工程成熟与否的重要标志。

3.4 生产环境的架构与参数建议

当RAG系统真正跑在服务上,需要考虑的问题和本地demo完全不是一回事。几个经验和建议:

第一是向量库的选型。Chroma在开发阶段很顺手,但是一旦数据量上到几十万条,或者需要多人同时写入,它就不太够用了。生产环境通常选用独立的向量数据库服务,比如Milvus、Qdrant、Elasticsearch(自带向量能力)。选型的核心指标是:写入吞吐、检索延迟、是否支持过滤条件、数据持久化策略。

第二是异步与缓存。大模型推理耗时在1到3秒之间,如果每个用户请求都完整地走一遍“向量检索+重排序+LLM生成”,后端压力会很大。我的做法是:检索结果做缓存,相同或相似的问题在小时级别内直接复用已经检索好的文档列表;生成结果做分级缓存,高频问题直接命中。直接从架构层面减少不必要的重推理。

第三是上下文窗口与计费策略。每次问答都要把检索到的文本拼进提示词,这部分token消耗看似不起眼,实际上在高并发下是一笔大开销。控制检索结果条数(通常4到6条就够)、控制每条文本长度(拼接前截断到512到1024字符),是既保证效果又控制成本的有效手段。很多时候把k=4改成k=3,答案质量不但没下降,反而更聚焦了。

4. 常见问题与排查实录

4.1 检索不到相关内容,答非所问

这是RAG最常被人吐槽的问题。排查路径按顺序走一遍:先用直接关键词在原始文档里搜索,确认问题答案确实在文档里。确认在的话,检查分块是否把答案切碎了——块与块之间的语义断裂会导致任何检索手段都命中不了。如果也没切碎,请检查Embedding模型是否适合你的领域语言风格。有些模型通用性好但专业术语效果一般,遇到这种情况要换模型。

一个典型案例:某次处理化工领域的检测报告,用通用中文模型做向量化,问“硫酸浓度检测上限”时始终检索不到关键段落。后来把Embedding模型换成在专业语料上微调过的版本,同样的代码直接命中。这个教训说明:当检索效果不好时,优先怀疑Embedding和文档处理的匹配度,而不是疯狂调参。

4.2 检索到了但答案仍然在编

检索质量正常,上下文就是正确答案,但大模型还是答偏。这种情况我遇到过几次,原因基本都是上下文组织的问题——提示词里给了模型太多自由度。对策是收紧提示词,明确要求“只能使用参考资料中的内容”,并且要求模型在信息不足时直接拒绝回答。同时检查重复信息是否过多:多条检索结果内容相似,等于变相减少了有效信息量。可以在拼接上下文之前做一次相似度去重。

4.3 系统回答慢

一个RAG请求涉及检索、向量化、重排序、大模型推理四个耗时环节,整体响应时间常常达到3到5秒。优化思路从三个方向同时进行:检索层用近似最近邻索引让向量检索降到毫秒级;生成层用更好的大模型换来更短的生成时间,或者用流式输出让用户感知延迟降低;架构层引入缓存,命中缓存的问题直接跳过检索和推理。

我实测过一个数据:加了上下文缓存之后,高频问题的响应从原来的3.8秒降到了0.7秒左右。缓存对RAG系统的性能收益是立竿见影的。

4.4 维护与成本:小步快跑优于一次性重建

文档更新是RAG上线后一直存在的需求。最粗放的做法是全部文档重新向量化,但当数据量变大后,这样做既慢又贵。更合理的做法是增量更新:文档新增时只处理新增部分,文档删除时通过元信息字段定位并清除旧向量。向量库只要支持基于metadata的过滤删除操作,这个流程很容易实现。我团队里的做法是每次更新前先在测试数据上跑一遍评估集,确认没有指标回退后再发布到生产。RAG本质上是数据在驱动,而数据的质量需要持续监控,没有任何一劳永逸的方案。

5. 写在最后的实践经验

做RAG项目这么长时间,我的体会是:它不是一个“调通API就完事”的demo型技术,而是一个需要反复迭代、不断用数据和真实文档来打磨的工程系统。很多人以为买一个向量数据库、套一个框架就是在做RAG,实际上真正见效果的关键,全部藏在文档分块的细节、检索排序的策略、提示词的约束这些不起眼的地方。

最后再分享一个很实用的小技巧:从一开始就把来源引用做到回答里。回答的末尾附上检索时命中的文档名称和片段,这样用户在业务场景里可以直接核对答案出处,也是在团队内部推动效果改进时最有说服力的证据。这个设计在任何RAG系统里都值得留一席之地。

RELATED READING

延伸阅读

更多一线实战笔记与深度复盘,助您持续精进