LangChain+RAG检索增强生成实战:从原理到生产级应用
简介这是一份面向AI开发者和研究者的LangChainRAG应用示例聚焦检索增强生成技术在智能问答、语义搜索等场景中的落地实践。项目以优质实战项目为蓝本完整演示了文档加载、文本切分、向量库构建、语义检索与生成回答的全链路流程并直观展示RAG如何借助外部知识源弥补大模型在专业知识和时效性上的不足。压缩包共含6个文件3个Python源码分别承担向量对比、数据库创建和查询交互功能2个Markdown文档提供环境搭建到调试优化的流程教程1个依赖清单便于复现环境整体仅65KB轻量完整。目前已有1400人学习下载适合具备基础Python能力、希望快速上手LangChain与RAG的开发者。通过边读教程边调试源码读者可掌握知识库构建、嵌入模型选择、检索参数调优等技巧并独立搭建出可运行的智能问答系统为后续在教育和客服等领域拓展应用打下基础。1. 一个基于 LangChain 与 RAG 的完整应用示例先搞懂检索增强生成再跑通源码做过大模型落地的同行都知道微调一个垂直领域模型成本有多高数据清洗、GPU 预算、迭代周期哪一样都能劝退小团队。而 RAG检索增强生成提供了一条更务实的技术路径让大模型先检索外部知识库再结合检索结果生成回答。这套基于 LangChain 构建的应用示例把索引构建、向量检索、上下文注入、流式输出这些核心环节全部串联起来附带的源码能直接改写成生产级项目。适合正在做知识库问答、私有化部署、法律文书辅助、客服工单自动回复这类场景的开发者也适合想系统理解 RAG 工程落地细节的技术负责人。关键是我拆完这个项目发现真正决定回答质量的不只是模型本身而是文档切块策略、Embedding 选型和召回参数的配合。2. RAG 架构拆解为什么检索增强能让大模型不再“睁眼说瞎话”2.1 从幻觉到有据可依RAG 的完整数据流大语言模型的知识有截止日期而且面对私有领域数据时它要么不知道要么一本正经地胡编。RAG 的思路很直接在模型生成前先从外部知识库检索出最相关的文档片段把这些片段和用户问题一起拼接成提示词再交给模型回答。这套流程让模型每次回答都有“参考资料”而不是凭记忆硬撑。我在实际拆解这个项目时先把数据流理了一遍文档加载从本地文件或目录读取原始文档支持文本、Markdown、PDF 等格式文档切块Chunking把长文档切成固定大小的片段设置重叠区域保证语义连贯向量化Embedding把每个文本片段转成稠密向量索引存储把向量存入向量数据库同时保留原文和元数据用户查询向量化把用户输入转为同一个向量空间的向量相似度检索在向量库中召回Top-K个最相关片段提示词拼接把检索到的片段注入System Prompt要求模型基于上下文回答生成回答模型按约束生成引用检索内容第一个让新手懵的点在步骤 2。许多人直接把整份文档丢进去结果召回的“相关片段”过于碎片化上下文被切得七零八落。项目里默认的切块大小是 500 个 Token重叠 50 个 Token这个默认值在处理技术文档时效果尚可但遇到表格和代码块时必须单独调整否则语义信息在切块处被截断。2.2 LangChain 在这个项目中的核心角色别把 Chain 当黑匣子这个项目选择 LangChain 而不是直接手写 OpenAI 调用和向量检索核心原因是 LangChain 把 RAG 流程封装成了一个可编排的链式结构。项目中实际上是 Chain 把文档加载器、文本分割器、向量化模型、检索器、提示词模板、大模型封装成一个完整的问答链路修改任何一个环节都只需要替换对应组件不需要重写整条流水线。LangChain 提供的关键抽象包括Document Loader统一的文档读取接口切换文件格式不用改业务代码Text Splitter可配的分割策略按 Token 数或按递归分隔符切分VectorStore向量存取抽象层可对接多种向量数据库Retriever从 VectorStore 派生支持相似度检索、MMR、带分数过滤的检索Prompt Template提示词模板用变量占位符拼接上下文LLM Chain组装上述组件的执行链项目源码里有一个模块专门负责解析文档另一个模块负责构建检索器。我看代码时发现一个小设计亮点检索器实例在应用启动时只创建一次之后每个请求复用同一个检索器实例而不是每次查询都重新加载整个索引。这个细节对生产环境部署非常关键直接决定接口的响应延迟是“秒级”还是“分钟级”。from langchain.schema import Document from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain.embeddings import OpenAIEmbeddings from langchain.vectorstores import FAISS def build_index(documents: list[Document], chunk_size: int 500, chunk_overlap: int 50) - FAISS: text_splitter RecursiveCharacterTextSplitter( chunk_sizechunk_size, chunk_overlapchunk_overlap, separators[\n\n, \n, 。, , , ] ) splits text_splitter.split_documents(documents) embeddings OpenAIEmbeddings() vectorstore FAISS.from_documents(splits, embeddings) return vectorstore这段代码中separators参数决定了切块的边界逻辑注意顺序是从大到小先按段落切再按句号切最后按空格兜底。如果文档以中文为主默认的[“\n\n”, “\n”, “ ”]会把大量中文句子拦腰截断所以我在实际使用中会把中文标点补充进分隔符列表。chunk_overlap的作用是让相邻片段有部分内容重叠避免语义被切断后下次检索完全命中不了。FAISS 应对百万级以内的文本量足够再往上量级才需要考虑换 Milvus 或 Qdrant。2.3 Embedding 选型思路OpenAI 还是本地模型项目骨架里默认用的是 OpenAI Embedding跑起来确实省事向量质量也稳定。但对接内部知识库时很多团队因为数据安全要求不能把文档发到云端就需要换本地 Embedding 模型。常见做法是切换成 HuggingFace 的 BGE 系列或国产的嵌入模型LangChain 提供了HuggingFaceEmbeddings封装改动量只在构建索引的那几行代码。这里有一个非常容易踩的坑Embedding 模型一旦确定后续所有查询和索引都必须沿用同一个模型。如果中途换模型向量空间直接错位检索结果就彻底失效。项目源码的配置项里把 Embedding 模型名写死了并没有做版本兼容如果你改完模型后发现召回结果异常第一反应应该是检查是不是新旧向量数据混着用了。# 切换本地 Embedding 模型的示例 from langchain.embeddings import HuggingFaceEmbeddings, OpenAIEmbeddings # 方式一使用 OpenAI Embedding项目默认 embeddings OpenAIEmbeddings(modeltext-embedding-ada-002) # 方式二使用开源本地模型数据不出内网 embeddings HuggingFaceEmbeddings( model_nameBAAI/bge-large-zh-v1.5, model_kwargs{device: cuda}, encode_kwargs{normalize_embeddings: True} )选 BGE 系列的主要原因是它对中文语义匹配效果好且 BGE 官方建议对向量做归一化处理后再计算相似度normalize_embeddingsTrue这个参数就是为此设置的。另外要注意 bge-large 模型参数量大约 400MCPU 推理时单个片段的向量化耗时可观如果用 CPU 跑完整索引建议换成 bge-small 或者调整批处理大小。3. 项目源码跑通实战从环境配置到命令行问答3.1 环境准备与依赖安装锁版本是第一要务下载源码后第一步是创建独立的虚拟环境避免把系统 Python 环境弄乱。项目要求的依赖集中在 requirements.txt 里LangChain 的版本迭代非常快我从这个项目里学到最重要的习惯就是严格锁定主依赖版本因为 LangChain 0.x 到 1.x 的 API 变了很多次不加锁依赖跑起来很不可控。# 创建并激活虚拟环境Windows / macOS / Linux 通用思路 python -m venv langchain-rag-env source langchain-rag-env/bin/activate # Windows 下使用 langchain-rag-env\Scripts\activate # 安装核心依赖。建议按固定版本安装避免最新版接口变动 pip install langchain0.1.0 langchain-openai0.0.5 langchain-community0.0.10 pip install faiss-cpu1.7.4 tiktoken0.5.2 pypdf3.17.4 python-dotenv1.0.0 # 如果计划使用 Chroma 替代 FAISS pip install chromadb0.4.22版本锁定这一段我是吃过亏的。之前在某模拟项目 X 中直接pip install langchain拉最新版结果上游依赖冲突pydantic版本不兼容启动时直接报导入错误。这个项目的 requirements.txt 里已经给你锁好了一组经过验证的版本组合运行前务必对照检查。如果后续要升级 LangChain 版本不要一次跳多个小版本每一步都跑一遍测试用例。3.2 配置文件与环境变量Key 放哪里决定你的代码能不能发给别人看项目根目录下有一个.env.example文件复制成.env后填入实际配置。这里涉及一个工程习惯的问题任何密钥都不应该硬编码进源码python-dotenv负责把.env文件中的键值对加载为环境变量.gitignore中忽略.env文件这样代码在 GitHub 上开源时才不会泄露密钥。# .env 示例内容 OPENAI_API_KEYsk-your-key-here EMBEDDING_MODELtext-embedding-ada-002 CHUNK_SIZE500 CHUNK_OVERLAP50 TOP_K4 VECTOR_STORE_PATH./storage/faiss_indexTOP_K是一个值得花时间调参的配置召回数量太少可能漏掉关键信息召回太多提示词被无关内容填充既浪费 Token 又容易把模型引向错误答案。我给 4 这个值算一个合理的起点后续根据实际回答质量再微调。VECTOR_STORE_PATH指定了向量索引持久化路径项目每次启动时先检查路径下是否有现成索引有就直接加载复用没有就全量构建。3.3 运行项目主流程三步走索引构建到问答闭环# 第一步准备知识文档 # 将待检索的文档支持 txt/md/pdf放入 ./knowledge 目录 # 第二步构建索引运行一次即可之后可反复复用 python build_index.py --source ./knowledge --output ./storage/faiss_index # 第三步启动交互式问答 python query.py第二步构建索引时控制台会输出切分的文本块数量、向量化耗时、索引存储路径。这里有玄学成分如果文档总量变化不大但索引构建速度越来越慢检查一下是不是每次都在向旧索引里追加数据而不是重建。项目源码里build_index.py的逻辑是先检查输出路径存在就加载旧索引并追加新文档这功能本身没毛病但旧索引若是用不同切块参数或不同 Embedding 模型生成的追加的数据会导致索引空间语义不一致。我习惯在更改算法参数后强制删除旧索引目录重建不冒追加的险。# query.py 核心代码骨架 from langchain.chains import RetrievalQA from langchain.llms import OpenAI # 加载已有索引 vectorstore FAISS.load_local( folder_pathVECTOR_STORE_PATH, embeddingsOpenAIEmbeddings(modelEMBEDDING_MODEL) ) # 构建检索器search_kwargs 控制召回数量和策略 retriever vectorstore.as_retriever( search_typesimilarity, search_kwargs{k: TOP_K} ) # 组装问答链verbose 打开后能观察到内部提示词拼接 qa_chain RetrievalQA.from_chain_type( llmOpenAI(temperature0), retrieverretriever, verboseTrue, chain_typestuff ) result qa_chain.run(项目文档里定义的超时阈值是多少) print(result)这个示例里chain_typestuff的含义是把召回的所有文档片段一次性全部塞进提示词。这种方式实现简单Token 占用可控适合片段数量少的场景。如果TOP_K超过 6 且文本块较大stuff 方式可能导致超 Token 限制那时就需要改用map_reduce或refine链类型。另外temperature0是问答场景的惯例设置减少随机性让回答更严谨。4. 进阶把示例改造成可用的知识库问答服务4.1 打造带引用的回答从黑匣子到可信输出项目自带的问答链直接返回大模型的文本输出没有标引用来源。真实业务场景里用户和审核方都希望知道回答依据的是哪些文档片段。改造方法是在上下文注入时保留每个片段的元数据在模型回答之后附加来源标识。LangChain 的RetrievalQA支持自定义combine_documents_chain我们可以把检索到的片段按文档名分组把引用信息随答案一起返回。from langchain.chains import RetrievalQAWithSourcesChain qa_chain RetrievalQAWithSourcesChain.from_chain_type( llmOpenAI(temperature0), retrieverretriever, return_source_documentsTrue, ) answer qa_chain(该系统的最大并发数是多少) print(answer[answer]) print(来源) for doc in answer[source_documents]: print(f - {doc.metadata.get(source, 未知文档)})source_documents里返回的是完整检索片段我在实战中会在后续增加一个过滤步骤按关键词把片段的所在页和大致段落位置提取出来单独展示。这一步看似多余实际上直接决定产品上线后用户信不信你。大模型输出的内容再准确没有来源标注就缺少说服力。4.2 流式输出解决“转圈等待”体验问题如果回答生成时间超过三秒用户就开始焦虑了。LangChain 支持通过回调处理器实现流式输出把模型逐 Token 生成的内容实时推送。改造后前端可以对接 Server-Sent Events 或者 WebSocket。这个项目源码中虽然没写完整的前后端交互但后端流式接口的骨架可以直接借鉴。from langchain.callbacks.base import BaseCallbackHandler class StreamHandler(BaseCallbackHandler): def __init__(self): self.text def on_llm_new_token(self, token: str, **kwargs) - None: self.text token print(token, end, flushTrue) handler StreamHandler() qa_chain RetrievalQA.from_chain_type( llmOpenAI(temperature0, streamingTrue), retrieverretriever, callbacks[handler] ) result qa_chain.run(简述项目中的限流策略)streamingTrue只是打开流式开关真正把增量文本推送出去需要配合回调函数。上边的示例把流式逻辑与问答流程解耦生产环境建议把StreamHandler.text改成回调事件通过消息队列把增量内容推给 Web 前端。还有一点是关于流式与夹具的配合使用流式输出时会话状态维护更复杂如果用了聊天历史和会话记忆功能要做好并发安全处理。4.3 测试验证与质量评估先量化再优化搭好了服务下一步就是用离线测试集评估回答质量。我习惯准备一批“问题-标准答案”对照集批量跑问答脚本分维度检查语义相关性回答是否切题事实准确性关键信息是否与标准答案一致引用正确性标注的来源是否能支撑结论# 批量测试脚本运行方式 python eval_qa.py --testset ./data/test_questions.json --output ./results/eval_report.json # 输出示例 # 测试样本200 条 # 相关率0.94 # 答案有效引用率0.81 # 平均回答长度137 字如果你评测时发现相关率不错但引用率偏低大概率是切块问题而不是模型问题。检索回来的块本身与问题相关但块中包含太多无用信息导致模型从无关段落摘录了内容。这时候优先调CHUNK_SIZE和CHUNK_OVERLAP这个项目的调参工具eval_qa.py会输出每一轮问答的检索片段 ID 和命中分数方便回溯定位。5. 避坑指南跑通这个 RAG 项目最容易翻车的地方5.1 现象FAISS 索引加载失败load_local直接报错现象重启服务后加载持久化索引提示维度不匹配或文件损坏。原因三种情况都能触发这个错误。第一种是更换了 Embedding 模型新文本向量维度和旧索引不一致第二种是 FAISS 索引文件被部分写入比如索引保存过程中程序被强制终止第三种是 LangChain 版本升级后FAISS 的序列化格式不兼容。解决删除storage/faiss_index目录重新跑build_index.py保证索引完整重建。如果插件停索引文件本身没有损坏排障时可以打印加载向量的维度与当前模型输出维度对照确认。从那以后每次改配置或升版本我都强制走一遍重建索引流程不再依赖增量追加。5.2 现象检索到的文本块完全不相干回答文不对题现象明明知识库里有关键信息回答却始终绕开重点或者直接说“未找到相关信息”。原因切块参数不合理导致语义断裂。如果你的文档是长表格或者代码片段500 Token 的切块策略会把完整逻辑切成两半检索时两边都只能召回一部分。另一个常见原因是分隔符列表中没有适配中文符号。解决观察verboseTrue模式下打印的检索片段如果每个片段只有一两句话说明切得太碎调大CHUNK_SIZE到 800 或 1000。如果片段切点落在句子中间在separators中补充中文句号、逗号让切块优先按完整句子边界断开。我做问答项目时第一件事往往是针对语料形态专门设计一套切块策略而不是直接用默认值。5.3 现象回答格式混乱模型把上下文当成格式化示例现象回答突发一堆 Markdown 表格、表情符号或奇怪的编号与设置的系统提示词风格完全不符。原因检索到的片段本身包含这些格式提示词里又没有明确“只输出纯文本”Llama 和 GPT 这类模型会把检索片段中的格式风格当真。解决在系统提示词里加一条硬约束比如“回答只输出纯文本不要使用 Markdown 表格或列表。如果上下文中有代码按代码块原样输出。”如果加约束还不管用检查知识库源文档里是否含有大量特殊字符在加载文档时做一次清洗。后面我把这个清洗环节也模块化了加载文档后统一过一遍规范过滤器。5.4 现象多个用户同时使用时内存暴涨服务直接卡死现象并发数一上来容器内存直线上升CPU 占满延迟飙升。原因问题出在检索器实例。每来一个请求代码里重新构建一次索引或加载一次模型资源开销被放大。另一个原因是 LangChain 的 LLM 实例每次请求重复创建底层连接没有复用。解决LangChain 的 LLM 实例和检索器在进程内是重量级对象应该在服务启动时实例化一次放进全局变量后续请求直接引用。向量数据库客户端使用单例模式LLM 调用层用连接池管理。这个项目源码里query.py没有做全局单例处理生产化时需要单独加一层工厂函数把索引加载、检索器创建和模型初始化包装到lru_cache或依赖注入容器里。5.5 现象中文标点被完全切碎检索结果命中率骤减现象输入中文问题返回片段经常是半个句子TOP1 片段相关性极低。原因默认的RecursiveCharacterTextSplitter分隔符对中文支持不够切完的块几乎每段都是一句话的碎片向量化后语义覆盖不够。解决自定义分隔符把“句号、问号、感叹号”加入数组我常用组合是[“\n\n”, “\n”, “。”, “”, “”, “”, “ ”, “”]。切完后额外检查是否有单字符块当出现频率过高时说明分隔符粒度太细适当删除底层分隔符。你也可以换用ChineseTextSplitter等适配中文语料的分割器但项目源码对依赖管理较严改动前先查兼容性。6. 部署与进阶技巧把示例项目推向生产环境的几个硬功夫6.1 服务化改造用 FastAPI 封装问答接口项目的query.py是命令行交互程序换成 Web 服务只需包一层 FastAPI。接口逻辑很简单接收请求体中的问题调检索链返回答案和来源。重要的是把初始化逻辑放到启动事件里避免第一个请求触发超长初始化。from fastapi import FastAPI from pydantic import BaseModel app FastAPI() qa_chain None class QueryRequest(BaseModel): question: str class QueryResponse(BaseModel): answer: str sources: list[str] app.on_event(startup) def load_model(): global qa_chain vectorstore FAISS.load_local(VECTOR_STORE_PATH, embeddings) qa_chain RetrievalQA.from_chain_type(llmOpenAI(temperature0), retrievervectorstore.as_retriever()) app.post(/ask, response_modelQueryResponse) def ask(req: QueryRequest): result qa_chain({query: req.question}) sources [doc.metadata.get(source, 未知) for doc in result[source_documents]] return QueryResponse(answerresult[result], sourcessources)这个接口骨架的关键在于“启动时加载模型”这一条很多人直接把它写在请求处理函数里结果每次接口调用都要等几十秒。用asyncio异步改造也可以但 RAG 链路中的检索部分是 CPU 密集或 IO 密集异步不一定能带来明显吞吐提升核心瓶颈通常在 Embedding 和 LLM 的耗时上。6.2 提示词调优的实操少一点“玄学”多一点工程验证很多人以为 RAG 回答质量由大模型能力决定实际测试后就会发现提示词模板对结果的影响可能超过模型本身。我反复调整项目里的 System Prompt总结了三个高价值技巧给出回答边界明确告诉模型“如果知识库中没有对应答案直接回复未收录不要自行推测”。这一句话可以把胡编乱造降一半。要求按点输出使用“请按 123 点列出结论每个结论后标注来源文档编号”。结构化输出不仅好读也方便程序解析。指定引用格式让模型在回答结尾列出“[来源文件名]”。这样后续处理引用来源时不需要再从大段文本中靠正则挖掘。prompt_template 你是知识库助手。请严格根据以下检索片段回答问题。 如果片段中没有答案请直接回复“知识库中未收录相关答案”不要编造。 检索片段 {context} 用户问题 {question} 要求 1. 只基于检索片段回答不要使用模型本身外部知识。 2. 回答按 1、2、3 列出要点每点后标注来源文档名。 3. 没有依据的信息不要输出。 回答 {context}和{question}这两个变量是 LangChain 提示词模板的关键占位符前者在运行时会被替换为检索到的文本块拼接结果后者替换为用户输入。我在实际使用中会额外在{context}中附带来源文件名作为元数据这样模型在标注来源时具备明确依据。这类微调不需要改代码逻辑只需要改一条提示词模板。6.3 一个稳妥的上线验证流程慢工出细活最后分享我在此类项目上的上线习惯。功能开发完不急着发布先找业务方要二十个真实问题跑一遍答案。再把回答发给业务方打分重点看“不知道就直接说不知道”的比例这个比例低说明模型在胡编是 RAG 提示词约束没做到位。解决完提示词问题再做一轮压力测试不断模拟多用户并发请求观察向量数据库连接消耗。我印象最深的一次是在某个内部工具上线前业务方问了 50 个问题模型有 6 个答案涉嫌编造来源排查到最后发现是检索召回阶段把两个高度相似的文档片段混在一起模型误把另一篇文章的结论当成了当前文档的内容。后来我调整了检索器的相似度阈值过滤掉置信度低于 0.75 的召回结果这类问题明显减少。从那以后每上线一个新知识域我都强制走一遍完整评估流程先测正确率再测并发两个指标都达标才发布。这套方法也值得你在复现这个项目后继续沿用希望帮到你。本文还有配套的精品资源点击获取