搞 RAG 知识库这件事LangChain 加国产大模型这套组合我前后折腾了将近一个月把能踩的坑基本都踩了一遍。从最开始只会调 API 返回固定格式到后面能跑通完整的“文档上传 - 切片 - 向量化 - 检索 - 生成回答”链路这中间的弯路确实不少。这篇文章就把我的完整实践过程拆开讲清楚从架构选型、环境配置、核心代码实现到 FastAPI 部署上线再到生产环境里那些绕不开的细节问题一次性说透。1. 做这个RAG知识库问答系统前先把架构建明白1.1 为什么选LangChain而不是自己手写流程RAG 的本质其实不复杂就是“先检索再生成”。用户问一个问题先从准备好的文档库里找出最相关的内容片段把这些片段拼进提示词里再丢给大模型让它基于这些材料作答。这个思路听着简单真做起来就会发现事情没那么轻松。文档格式五花八门有 PDF、Word、Markdown、HTML光是解析和清洗文本就够写几百行代码。切分策略要反复调试向量数据库的选型和索引参数也有讲究更别提检索结果怎么排序、怎么和生成环节衔接这些问题。我一开始也想自己写毕竟 RAG 的流程不算高深用 requests 库调 API 加上一个向量库客户端也能拼出来。但写着写着就发现自己造轮子的成本远高于预期。光是“文档加载器”这一块LangChain 已经封装了上百种格式的解析器PDF 有 PyPDFLoaderWord 有 Docx2txtLoaderMarkdown 有 UnstructuredMarkdownLoader每种都处理了编码、表格、页眉页脚这些烦人的细节。另外像 Prompt 模板、输出解析器、链式调用这些都做得非常成熟。还有一个很现实的原因——团队协作和后期的维护成本。自己用手写代码拼的流程逻辑分散在各个文件里后来接手的人要看半天才能看懂。LangChain 做的事情就是把这些组件标准化每个环节都是一个可替换的模块新人上手快出了问题排查也方便。说白了LangChain 在这个项目里扮演的角色就是“装配车间”把文档处理、向量检索、模型调用这些零件按顺序组装起来。1.2 为什么必须用向量数据库以及哪款适合你RAG 的核心检索逻辑是“语义相似度检索”也就是把用户的问题变成一组向量然后去向量数据库里找和这组向量最接近的文本片段。传统数据库做不了这件事因为它是基于关键词和精确匹配的你说“怎么申请年假”它找不到文档里写着的“休假管理规定”。所以向量数据库是 RAG 系统里绕不开的一个组件。市面上的选择不少Chroma 适合个人项目跑原型Milvus 适合大数据量和高并发的场景PGVector 适合不想额外维护一个数据库服务的团队因为可以直接利用现有的 PostgreSQL 实例。我在这个项目里用了 Chroma 做开发调试后期迁移到生产环境时换成了 PGVector文章里会把两种方案的部署方式和切换方法都写清楚。先说说我的选型标准。数据量在百万级文档以内QPS 要求不高直接上 Chroma 就够了它的零配置特性让开发效率提升非常明显。但如果是企业级应用需要对接已有的权限体系或者对查询延迟有硬性要求那就得考虑 PGVector 或 Milvus。还有一个容易忽略的点——向量数据库的备份和恢复机制。Chroma 的持久化目录复制到另一台机器上就能用但这个特性同时也是隐患并发写入容易出现数据损坏。PGVector 就没这个问题数据由 PostgreSQL 统一管理事务和备份都很成熟。1.3 国产大模型怎么接API方案和本地部署怎么选国产大模型这几年的API服务已经非常成熟了国内常见的几款——通义千问qwen系列、DeepSeek、智谱清言的GLM系列、Kimimoonshot——都提供了兼容OpenAI格式的接口。这意味着你只需要把base_url换成对应的地址把api_key换成自己的密钥其余代码几乎不用改。选择 API 方案还是本地部署取决于你的业务场景。API 方案的优势是启动快、免运维、按量付费适合中小型项目快速验证本地部署的优势是数据不出内网、隐私性更强、长期使用成本可控但需要准备显卡资源。以通义千问的 qwen-plus 模型为例API 调用延迟在 2 秒左右上下文窗口足够容纳我们的检索结果在知识库问答场景下表现是比较均衡的。如果选择本地部署目前比较主流的方案是 Ollama 加 Qwen 或 GLM 的开源版本。比如 Qwen2.5-7B-Instruct 在消费级显卡上就能跑得动量化后显存占用控制在 8GB 左右我用一台 3060 12G 的机器实测生成速度大概在每秒 15 到 20 个 token做内部知识库问答够用了。DeepSeek 的开源模型比如 DeepSeek-V3性能更强但对显存的要求也高了不少一般是团队有 A100/H100 级别资源才会考虑。2. 环境准备与依赖安装2.1 Python版本与虚拟环境避坑指南这个项目我用了 Python 3.10这个版本兼容性最稳。Python 3.11 和 3.12 也能跑但有些库比如 onnxruntime 的某些版本在 Windows 上可能找不到预编译的 wheel 包装起来比较折腾。虚拟环境一定要建别图省事直接往全局环境里装。我见过太多人把 langchain、torch、numpy 这些包直接 pip install 到系统 Python 里装到一半发现和已有项目的依赖冲突要么卸载重装要么破罐子破摔。用 venv 和 conda 都可以我个人习惯用 conda因为 Python 版本切换比较方便。提示在 Windows 上创建 conda 环境的时候指定 Python 3.10 这个版本最稳妥。2.2 安装核心依赖包清单及版本说明核心依赖就这几个langchain框架主体langchain-community包含了各种文档加载器和向量库适配器langchain-openai用来对接兼容 OpenAI 格式的国产大模型 APIchromadb本地向量数据库fastapi 和 uvicorn部署 API 服务python-dotenv管理 API 密钥pypdf 和 docx2txt解析 PDF 和 Word 文档用tiktoken计算 token 数量用来控制切分长度安装命令pip install langchain langchain-community langchain-openai chromadb pip install fastapi uvicorn python-dotenv pip install pypdf docx2txt tiktoken这里要说一个坑langchain 这个库从 0.1.0 版本开始做了大规模的模块拆分以前一个包包含所有东西现在变成了 langchain 只有核心的链逻辑文档加载器和各种第三方集成全部分散到 langchain-community 里。如果你直接from langchain.document_loaders import PyPDFLoader在新版本里会报错正确写法是from langchain_community.document_loaders import PyPDFLoader。2.3 获取国产大模型API Key的详细流程以通义千问为例。去阿里云百炼控制台注册账号完成实名认证然后在模型广场找到 qwen-plus 模型点击“开通服务”并创建 API-KEY。这个过程大概十分钟就能搞定新用户一般都有免费额度做实验绰绰有余。DeepSeek 就在开放平台注册账号创建 API Key 就行充值最低 10 块钱够跑上百次问答了。拿到 API Key 之后在项目根目录创建一个.env文件DASHSCOPE_API_KEY你的通义千问key DEEPSEEK_API_KEY你的DeepSeek key然后写一个工具函数来加载环境变量from dotenv import load_dotenv load_dotenv()需要说明的是LangChain 对接国产大模型有两种方式。如果模型厂商提供了兼容 OpenAI 的接口直接用ChatOpenAI指定base_url就行如果厂商自己的 SDK 封装得比较好也可以继承 LangChain 的BaseChatModel类做定制封装。我后面代码里用的是第一种方式简单直接。2.4 Embedding模型的选用与部署细节Embedding 模型这一步很多人会忽略。其实 RAG 系统的回答质量瓶颈大多数时候不在大模型而在检索环节。检索的效果取决于两件事切分策略合不合理以及 Embedding 模型的语义理解能力。国内使用比较友好的是阿里云的 text-embedding-v3 系列通过 API 调用效果不错。如果不想用 API有个偏轻量的开源方案是 BAAI/bge-small-zh-v1.5这个模型只有 24M 参数CPU 上就能跑对中文的支持和维度控制都做得比较好。我项目里用的是 API 方案调用方式from langchain_openai import OpenAIEmbeddings embeddings OpenAIEmbeddings( modeltext-embedding-v3, base_urlhttps://dashscope.aliyuncs.com/compatible-mode/v1, api_keyos.getenv(DASHSCOPE_API_KEY) )Embedding 的维度也值得关注。text-embedding-v3 默认是 1024 维bge-small-zh-v1.5 是 512 维。维度越高代表每个文本的语义特征越丰富检索的精细度也会更好但相应的存储空间和计算耗时都会增加。在数据量不是特别大的场景下512 维和 1024 维的差距并不明显。选模型的时候还是优先看语义匹配的效果单看数字意义不大。3. 核心实现从文档加载到问答输出3.1 文档加载与解析的完整处理流程文档加载是整个 RAG 流程的入口也最容易踩坑。PDF 文件的解析是个老大难问题扫描版的 PDF 需要 OCR文字版的 PDF 还会遇到表格错乱、页眉页脚混入正文的情况。我实测下来PyPDFLoader 对文字版 PDF 的解析效果尚可但遇到分栏排版或者表格内容较多的文档结构会丢失得比较厉害。相关代码from langchain_community.document_loaders import PyPDFLoader from langchain_community.document_loaders import Docx2txtLoader from langchain_community.document_loaders import TextLoader from langchain.text_splitter import RecursiveCharacterTextSplitter def load_document(file_path): if file_path.endswith(.pdf): loader PyPDFLoader(file_path) elif file_path.endswith(.docx): loader Docx2txtLoader(file_path) else: loader TextLoader(file_path, encodingutf-8) return loader.load()一个实际经验很多企业内部知识库的文档都是 PDF 格式但 PDF 的排版没有统一标准有的带目录、有的带水印、有的页脚是公司名。如果你在加载之后直接做切分页脚的水印文本会混进知识库检索的时候动不动就把这些无关内容匹配出来。所以文档清洗这一步很重要我一般在加载之后会做一次正则过滤把明显无关的内容剔除掉。注意不要指望 LangChain 自带的 loader 能完美处理所有 PDF。如果实测下来解析效果不理想建议先用 pypdf 或者 pdfplumber 做预处理把内容按“阅读顺序”重新整理好再交给 LangChain。我遇到分栏的 PDF 时会用 pdfplumber 手动提取坐标和文本然后按坐标顺序重排虽然代码多写了三五十行但检索准确率的提升是肉眼可见的。3.2 文本切分策略的核心参数讲解文本切分是 RAG 里最被低估的环节。切分的粒度直接决定了检索的“命中精度”。如果切分得太粗比如把一个 50 页的文档整个丢进向量库那检索出来的是一个超大文档里面相关内容只占很小一部分大模型读了大部分无关内容回答的质量自然就差了。如果切分得太细比如按一句话一切那检索出来的片段缺少上下文大模型看到的信息是孤立的也没办法给出准确回答。我实测了近十种切分策略最稳定的是 RecursiveCharacterTextSplitter。它的原理是递归地按照一系列分隔符来切分文本优先保留语义完整的段落和句子。参数设置如下from langchain.text_splitter import RecursiveCharacterTextSplitter text_splitter RecursiveCharacterTextSplitter( chunk_size500, chunk_overlap50, separators[\n\n, \n, 。, , , ., !, ?, , ;, , ,, , ] )chunk_size 是每个片段的长度上限按字符数计算chunk_overlap 是相邻片段之间的重叠长度。两个参数的核心逻辑是既要保证单个片段有足够的上下文信息又不能超出 Embedding 模型的最大输入长度和检索匹配的精度范围。选择 500 这个值是因为我测过大概 300 到 800 这个区间300 时片段太碎检索出的内容经常答非所问800 时片段太长容易混入过多无关信息500 是一个比较平衡的点。chunk_overlap 设为 50 是为了避免一个完整句子被从中间截断让前后两个片段各保留一点交叉信息这样即使需要跨片段的信息检索也能命中。还有一个容易被忽略的问题chunk_size 的计量单位。LangChain 的 RecursiveCharacterTextSplitter 默认按字符数计算对中文来说这个方式基本够用。但如果你的文档是中英混排建议换用 TokenTextSplitter按 token 数来切分这样对模型的提示词长度控制更精确。切分完成之后还有一个可选但强烈建议的步骤——用 metadata 标记每个片段的来源信息包括文件名、页码、章节标题。这样检索到结果后系统能告诉用户“这个答案来自《员工手册》第 3 页”可信度和使用体验会好很多。3.3 向量化与知识库入库的实现方式把所有切分好的文本片段交给 Embedding 模型生成向量然后写入向量数据库。这个过程在 LangChain 里一行代码就能完成from langchain_community.vectorstores import Chroma vectorstore Chroma.from_documents( documentsdocs, embeddingembeddings, persist_directory./data/chroma_db )这里有个性能问题值得聊一聊。Chroma.from_documents是逐条向量化再写入的如果文档量很大比如几千个 PDF这个过程会非常慢。我做过一个 2000 个 PDF 的导入实验用默认方式跑了一晚上才完成。后来改用批处理方式你自己把文本分批传给 Embedding 接口然后把生成的向量用Chroma.add_embeddings批量写入时间直接缩短到原来的三分之一。核心是 Embedding API 的并发请求限制不同的服务商对 QPS 的限制不同实际使用的时候要做一层并发控制既不能太慢也不要把 API 打挂。入库之后别忘了做一次简单的验证。从库里随机抽几个问题用相似度检索看能不能返回相关的片段。这一步能帮你判断前面的切分和向量化效果不要等整个系统搭完了才发现知识库压根检索不到正确内容。3.4 检索增强与问答链实现代码详解系统走到这一步开始做真正的“检索增强”。LangChain 提供了一个叫 RetrievalQA 的链把所有环节串起来from langchain.chains import RetrievalQA from langchain_openai import ChatOpenAI llm ChatOpenAI( modelqwen-plus, base_urlhttps://dashscope.aliyuncs.com/compatible-mode/v1, api_keyos.getenv(DASHSCOPE_API_KEY), temperature0.3 ) qa_chain RetrievalQA.from_chain_type( llmllm, chain_typestuff, retrievervectorstore.as_retriever(search_kwargs{k: 4}), return_source_documentsTrue )这段代码里最值得斟酌的是chain_typestuff和search_kwargs{k: 4}这两个参数。stuff 模式是把检索到的所有片段直接拼接到提示词里适合片段数量少一般不超过 5 个且总长度在模型上下文窗口内的场景。如果你的知识库结构复杂、片段数量多可以考虑 map_reduce 或 refine 模式。map_reduce 会对每个片段单独提问再汇总答案效果更稳定但调用成本翻倍。refine 则是在已有答案的基础上逐步修正适合迭代优化场景。实际项目里我建议先从 stuff 开始如果回答质量不达标再上 refine。k4表示返回最相似的 4 个片段。这个值不是越大越好片段太多会造成两个问题一是提示词过长模型可能混淆重点二是无关信息可能被噪声干扰反而降低了回答的准确性。我试过 k 从 2 到 8 的区间4 到 5 是效果和成本的平衡点。再来看 Prompt 的设计。默认的 Prompt 是英文模板直接用在中文场景会出现表达风格漂移模型会用英文答辩式的口吻回答中文问题。所以我在实际项目中自定义了中文 Promptfrom langchain_core.prompts import PromptTemplate prompt_template 你是一个知识库问答助手请根据以下已知信息用中文简洁、准确回答用户的问题。 已知信息 {context} 用户问题{question} 回答要求 1. 如果已知信息中有相关内容请基于这些内容进行回答并注明内容来源。 2. 如果已知信息中没有相关内容请直接回答“知识库中没有找到相关内容”禁止编造。 3. 回答要条理清晰重点突出。 回答 QA_CHAIN_PROMPT PromptTemplate( templateprompt_template, input_variables[context, question] ) qa_chain RetrievalQA.from_chain_type( llmllm, chain_typestuff, retrievervectorstore.as_retriever(search_kwargs{k: 4}), return_source_documentsTrue, chain_type_kwargs{prompt: QA_CHAIN_PROMPT} )这个 Prompt 模板里面“禁止编造”这条约束在 RAG 系统里非常关键。一个常见的误区是完全信任大模型不做任何约束。实际上大模型在知识库没有相关内容时很容易“一本正经地胡说八道”把这个禁止条款写进提示词能显著减少这种幻觉。3.5 完整代码串联与使用效果展示把所有环节组装起来完整的问答函数也就几十行代码import os from dotenv import load_dotenv from langchain_community.document_loaders import PyPDFLoader, Docx2txtLoader from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain_openai import ChatOpenAI, OpenAIEmbeddings from langchain_community.vectorstores import Chroma from langchain.chains import RetrievalQA from langchain_core.prompts import PromptTemplate load_dotenv() embeddings OpenAIEmbeddings( modeltext-embedding-v3, base_urlhttps://dashscope.aliyuncs.com/compatible-mode/v1, api_keyos.getenv(DASHSCOPE_API_KEY) ) CHROMA_PATH ./data/chroma_db def build_knowledge_base(file_paths): all_docs [] for fp in file_paths: if fp.endswith(.pdf): loader PyPDFLoader(fp) elif fp.endswith(.docx): loader Docx2txtLoader(fp) else: continue all_docs.extend(loader.load()) splitter RecursiveCharacterTextSplitter( chunk_size500, chunk_overlap50, separators[\n\n, \n, 。, , , ., !, ?, , ;, , ,, , ] ) docs splitter.split_documents(all_docs) vectorstore Chroma.from_documents( documentsdocs, embeddingembeddings, persist_directoryCHROMA_PATH ) return vectorstore def create_qa_chain(vectorstore): llm ChatOpenAI( modelqwen-plus, base_urlhttps://dashscope.aliyuncs.com/compatible-mode/v1, api_keyos.getenv(DASHSCOPE_API_KEY), temperature0.3 ) prompt_template 你是一个知识库问答助手请根据以下已知信息用中文简洁、准确回答用户的问题。 已知信息 {context} 用户问题{question} 回答要求 1. 如果已知信息中有相关内容请基于这些内容进行回答并注明内容来源。 2. 如果已知信息中没有相关内容请直接回答“知识库中没有找到相关内容”禁止编造。 3. 回答要条理清晰重点突出。 回答 QA_CHAIN_PROMPT PromptTemplate( templateprompt_template, input_variables[context, question] ) return RetrievalQA.from_chain_type( llmllm, chain_typestuff, retrievervectorstore.as_retriever(search_kwargs{k: 4}), return_source_documentsTrue, chain_type_kwargs{prompt: QA_CHAIN_PROMPT} ) if __name__ __main__: vs build_knowledge_base([./data/handbook.pdf]) chain create_qa_chain(vs) result chain.invoke({query: 年假可以分几次休}) print(回答:, result[result]) print(来源:, [doc.metadata.get(source) for doc in result[source_documents]])这版代码在本地跑通之后我用一份 50 页的员工手册做了测试问了几个问题比如“带薪年假的享受条件是什么”“薪资发放日期是哪天”回答都能从文档中找到对应内容而且标注了来源页数。当然也遇到过答非所问的情况比如把产品说明书里的术语和员工手册里的内容混淆了这就要靠后面的混合检索和重排序来优化。4. 用FastAPI把RAG服务暴露成HTTP接口4.1 项目目录结构与FastAPI应用搭建前面写的代码在命令行里跑通了但知识库问答系统的价值要发挥出来必须提供一个 API 服务让前端页面、企业微信机器人、钉钉应用等都能调用。我把项目拆成了模块化的结构rag_project/ ├── .env ├── requirements.txt ├── app.py # FastAPI 入口 ├── rag/ │ ├── __init__.py │ ├── loader.py # 文档加载 │ ├── splitter.py # 文本切分 │ ├── vectorstore.py # 向量库操作 │ ├── qa_chain.py # 问答链 │ └── config.py # 全局配置 ├── data/ │ └── kb_docs/ # 知识库原始文档 └── vectorstore/ └── chroma_db/ # 向量数据库持久化目录FastAPI 这个框架接口很直观一个基础服务大概长这样from fastapi import FastAPI, UploadFile, File, HTTPException from pydantic import BaseModel app FastAPI(titleRAG 知识库问答系统) class QueryRequest(BaseModel): question: str top_k: int 4 class QueryResponse(BaseModel): answer: str sources: list[str] app.post(/qa, response_modelQueryResponse) async def answer_question(req: QueryRequest): try: result qa_chain.invoke({query: req.question}) return QueryResponse( answerresult[result], sources[doc.metadata.get(source, ) for doc in result[source_documents]] ) except Exception as e: raise HTTPException(status_code500, detailstr(e)) app.post(/upload) async def upload_document(file: UploadFile File(...)): # 保存文件并更新知识库 ...4.2 接口设计精髓与文档上传功能实现真正上生产接口设计要考虑的细节远比上面这个 demo 多。/qa接口做问答/upload接口做文档上传这两个接口的职责要分开因为文档上传后的向量化入库是异步耗时的操作不能阻塞问答接口的响应。我的做法是/upload接口接收文件后先保存到磁盘返回一个document_id把“文档解析 - 切分 - 向量化 - 入库”这个流程丢到后台任务队列里执行/qa接口查询的时候只从已经完成入库的向量库里检索文档上传功能的实现import uuid from pathlib import Path from fastapi import BackgroundTasks UPLOAD_DIR Path(./data/kb_docs) UPLOAD_DIR.mkdir(exist_okTrue) app.post(/upload) async def upload_document(file: UploadFile File(...), background_tasks: BackgroundTasks None): doc_id str(uuid.uuid4()) suffix Path(file.filename).suffix save_path UPLOAD_DIR / f{doc_id}{suffix} with open(save_path, wb) as f: content await file.read() f.write(content) background_tasks.add_task(process_document, str(save_path), doc_id) return {document_id: doc_id, status: processing} def process_document(file_path: str, doc_id: str): try: docs load_document(file_path) splitter RecursiveCharacterTextSplitter( chunk_size500, chunk_overlap50, separators[\n\n, \n, 。, , , ., !, ?, , ;, , ,, , ] ) chunks splitter.split_documents(docs) for chunk in chunks: chunk.metadata[doc_id] doc_id vectorstore get_vectorstore() vectorstore.add_documents(chunks) vectorstore.persist() except Exception as e: print(f处理文档失败: {e})因为这个接口在同一时间可能被多个用户调用向量数据库的并发写入需要特别小心。Chroma 在并发写入时容易出问题如果服务端有多个 worker建议在写入前加一个线程锁或者直接换成 PGVector。4.3 登录鉴权与请求日志生产环境的几个关键点这一步属于生产环境的基本要求了没有鉴权的 API 相当于把知识库完全开放哪怕只是内网使用风险也很大。鉴权方案我用的是最简单有效的 API Key 机制。在.env里配置一个API_AUTH_TOKEN然后在 FastAPI 里加一个依赖函数from fastapi import Depends, Header API_AUTH_TOKEN os.getenv(API_AUTH_TOKEN, secret-token) async def verify_token(authorization: str Header(...)): if authorization ! fBearer {API_AUTH_TOKEN}: raise HTTPException(status_code401, detail认证失败) app.post(/qa, response_modelQueryResponse, dependencies[Depends(verify_token)]) async def answer_question(req: QueryRequest): ...请求日志这一块也别省。每次问答请求记下问题内容、回答内容、检索到的来源文档、响应时长和状态码。日志累积之后可以用来分析用户的提问分布比如哪些问题是知识库没覆盖到的哪些文档经常被检索到这些都是优化知识库的宝贵数据。4.4 启动服务的正确姿势与调试技巧启动 FastAPI 服务官方推荐用 uvicornuvicorn app:app --host 0.0.0.0 --port 8000开发调试的时候记得加--reloaduvicorn app:app --host 0.0.0.0 --port 8000 --reload生产环境部署建议加几个参数--workers 4配合 Gunicorn 做多进程管理--timeout 120防止长请求超时被误杀。我一个实际经验是大模型生成速度如果不是流式接口有可能超过默认的 30 秒超时设置。要么把超时设大要么改造成流式输出。流式输出对用户体验的提升非常明显建议优先做这件事。注意多进程模式下Chroma 持久化目录的并发访问有损坏风险。稳妥方案是把知识库存储切换到 PostgreSQLPGVector或者用 Redis 做一层分布式锁。如果数据量不大直接单进程部署也可以接受但要有明确的容量规划。5. 生产级RAG系统优化检索质量与性能调优5.1 混合检索策略向量检索与关键词检索的融合纯向量检索有个明显短板——对专有名词、产品型号、员工编号这类“精确信息”的支持不够好。比如用户问“型号 ABC-100 的功率是多少”向量检索可能把它语义扩展到“设备参数”“运行效率”反而忽略了最关键的型号字符串。这时候就需要引入关键词检索来做精确匹配。我在项目里实现了一套混合检索逻辑向量检索负责语义召回BM25 负责精确关键词匹配两者取并集再通过分数融合排序。LangChain 提供了EnsembleRetriever来组合各种检索器from langchain.retrievers import BM25Retriever, EnsembleRetriever from langchain_community.vectorstores import Chroma vector_retriever vectorstore.as_retriever(search_kwargs{k: 4}) bm25_retriever BM25Retriever.from_documents(chunks, k3) ensemble_retriever EnsembleRetriever( retrievers[bm25_retriever, vector_retriever], weights[0.4, 0.6] ) qa_chain RetrievalQA.from_chain_type( llmllm, chain_typestuff, retrieverensemble_retriever, return_source_documentsTrue, chain_type_kwargs{prompt: QA_CHAIN_PROMPT} )权重系数 0.4 和 0.6 是我实测后定的。BM25 权重太高会让回答变得“死板”只按字面匹配向量检索权重太高又丢掉了精确匹配的优势。这个比例在不同领域的知识库上表现有差异工程实践中需要根据自己的数据做小规模的调参测试。5.2 重排序Rerank机制提升答案准确率的关键一步初检阶段拿到的片段往往有噪声比如某个片段同时包含了好几个主题的内容而用户问的只是其中一个。这时候如果把这些混杂的片段直接拼给大模型回答的准确性会受影响。解决方案是加一个重排序层用一个更强的模型对初检结果进行精细打分把最相关的片段排到最前面。我用的方式是调用一个独立的 Rerank API 或者本地部署一个 rerank 模型比如 bge-reranker。处理逻辑是初检拿到 10 个候选片段Reranker 逐条计算与问题的相关度分数最后只取分数最高的 4 个喂给大模型。这一层“粗排 精排”的思路在搜索系统里很成熟用到 RAG 里同样效果显著。我实测过加了 Rerank 之后回答准确率大概提升了 15% 到 20%尤其是在知识库内容接近、模糊问题较多的场景下效果提升很明显。代价是查询响应时间增加了大概 300 到 500 毫秒但换来的准确率提升完全值得。5.3 缓存机制用语义缓存降低重复请求开销知识库问答场景有一个特点很多问题会被反复提问。比如“怎么修改密码”这个问题几十个员工会在一周内反复问。每次都对相同问题做完整检索和生成浪费计算资源不说响应时间也不稳定。解决方案是做一层语义缓存。我的实现思路是把用户的问题向量化在缓存区里做相似度匹配如果找到相似度超过 0.95 的历史问题就直接返回历史答案不再走大模型。这个方案是语义级别的缓存和传统对输入文本做精确匹配的缓存不同。class SemanticCache: def __init__(self, max_len1000): self.cache [] self.max_len max_len def get(self, question_vector): if not self.cache: return None # 简单循环比对生产环境建议用向量数据库实现 for cached_qv, answer in self.cache: similarity cosine_similarity(question_vector, cached_qv) if similarity 0.95: return answer return None def put(self, question_vector, answer): self.cache.append((question_vector, answer)) if len(self.cache) self.max_len: self.cache.pop(0)实测效果非常可观。知识库问答系统上线后前两周的缓存命中率大约在 30% 到 40%基础问题的响应时间直接降到 100 毫秒以内。这里要注意缓存的过期策略——知识库更新之后旧缓存需要主动失效否则用户会拿到过时信息。5.4 提示词压缩技术让回答更精准不跑偏检索到的片段里真正和问题强相关的内容往往只占一部分。有时候我们把 4 个片段拼进提示词其中 2 个完全不相关这会让大模型在生成回答时被干扰。解决这个问题除了 Rerank还可以做提示词压缩。LangChain 有个ContextualCompressionRetriever的机制。它让一个大模型或者 reranker 先看片段内容和用户问题把片段中与问题不相关的部分过滤掉只保留最核心的文本再进入最终提示词。这样做的好处是大幅减少了提示词中的噪声回答时模型可以更聚焦。from langchain.retrievers import ContextualCompressionRetriever compression_retriever ContextualCompressionRetriever( base_compressorreranker_model, base_retrieverensemble_retriever )理论上这层会再次增加延迟因为每个片段都要过一遍压缩模型。但在追求高质量回答的场景中这层是最有效的“提纯”手段。我一般是按需开启如果知识库内容噪声大、检索结果经常跑偏就开启如果知识库本身质量高、用户问答也比较集中这层可以不开。6. 常见问题与排查技巧实录6.1 高频报错整理与解决方案速查表调试过程中遇到最多的问题我用一个表格整理出来方便大家对照排查报错信息原因分析解决方案ModuleNotFoundError: No module named langchain_community安装的 LangChain 版本过新文档加载器已拆分为独立包pip install langchain-community改从langchain_community导入OpenAIError: Invalid base_urlbase_url 拼写有误或者漏掉了兼容模式路径检查是否写成https://dashscope.aliyuncs.com/compatible-mode/v1完整地址RateLimitError调用国产大模型 API 频率过高加入指数退避重试机制或者调整并发数chromadb.errors.UniqueConstraintError重复添加了相同 ID 的文档片段写入前检查doc_id是否冲突使用uuid4确保唯一性FastAPI 接口超时同步调用大模型生成耗时长于默认超时设置改为流式响应或调大 uvicorn 的timeout检索不到任何内容向量库为空或 Embedding 维度不匹配检查向量库是否有数据确认embedding模型是否一致回答内容与知识库无关chunk_size 过大或过小导致语义不聚焦调整切分参数为 500/50增加混合检索和 RerankPyPDFLoader 解析乱码PDF 是扫描版或使用了特殊编码改用 OCR 工具预处理或使用 pdfplumber 提取6.2 踩坑实录几个让人印象深刻的调试经历第一个坑是 Embedding 模型不一致导致的“检索不到内容”。我在开发机用 bge-small-zh-v1.5 生成了向量并存入 Chroma然后部署到服务器上时换了 text-embedding-v3结果所有查询都返回空结果。排查了很久最后发现两个模型生成的向量维度不一致512 维和 1024 维相似度计算全部失效。这个问题的教训是从开发到生产的整个链路里Embedding 模型必须一致最好在配置文件中写死不能靠人工记忆。第二个坑是 PDF 表格解析的失真。我用 PyPDFLoader 加载一份带表格的技术参数文档切分后的文本完全乱掉了表格的列标题和数据行被拆分到不同片段检索出来给大模型一解读就全错了。后来我改用 pdfplumber 专门处理这类表格先提取表格结构再按行转成文本效果好了很多。遇到大段文字加表格混排的文档时我会单独写一个预处理函数先按坐标和类型段落、表格、页眉页脚分区域提取再按阅读顺序重组。第三个坑是 FastAPI 多进程模式下 Chroma 的锁冲突。我在生产环境用 Gunicorn 起了 4 个 worker结果上传文档和问答同时进行时Chroma 经常报database or disk is full或者直接卡死。查了一圈发现是 SQLite 后端的并发写入问题。后来我把向量库换成了 PGVector问题就彻底解决了。如果数据量和 QPS 不大也可以用单 worker 线程锁的方式应对。6.3 新模型库跑通后的一些优化心得整套系统跑通之后我觉得从“能用”到“好用”还有三件事值得做。第一件事是知识库的更新机制。企业里的文档不是一成不变的制度会更新产品说明会修订。RAG 系统要有一个清晰的知识版本管理方案——上传新文档时旧版本的向量是保留还是删除文档内容变了之前缓存的答案要不要清理我的做法是给每个片段打上文档版本号更新时按doc_id删除旧向量再写入新向量缓存按周期自动失效。第二件事是回复的可解释性。在给用户的回答中标注来源文档和具体页码这个体验在内部系统中尤其重要。员工看到答案来自哪份文件信任度会高很多。实现上已经做了return_source_documentsTrue就能拿到来源关键是把这些信息展示到前端界面上。第三件事是持续的评测与优化。我建了一个包含 200 条测试问题的评估集每次调整切分参数、检索策略、Prompt 模板之后都要跑一遍回归测试对比回答准确率的变化。没有这个评估集所有优化都是“凭感觉”很难判断某个改动到底改善了还是劣化了。这个过程虽然枯燥但它是让知识库问答系统走向生产级的必经之路。最后再聊一点个人体会。做这类系统别一上来就追求“大而全”。先用最简单的方式把问答链路跑通用一个小文档库验证效果然后一个模块一个模块地往里加优化——先加混合检索再加 Rerank再上缓存每一步都做对比评测。这样迭代方向清晰出现问题时也能快速定位。我见过太多人一开始就把 Milvus、LangGraph、Agent 这些概念堆上去结果系统复杂到出了问题根本不知道往哪里查。技术栈再先进能解决问题、稳定上线才是硬道理。 LangChain+国产大模型：RAG知识库问答系统从零到生产部署
搞 RAG 知识库这件事LangChain 加国产大模型这套组合我前后折腾了将近一个月把能踩的坑基本都踩了一遍。从最开始只会调 API 返回固定格式到后面能跑通完整的“文档上传 - 切片 - 向量化 - 检索 - 生成回答”链路这中间的弯路确实不少。这篇文章就把我的完整实践过程拆开讲清楚从架构选型、环境配置、核心代码实现到 FastAPI 部署上线再到生产环境里那些绕不开的细节问题一次性说透。1. 做这个RAG知识库问答系统前先把架构建明白1.1 为什么选LangChain而不是自己手写流程RAG 的本质其实不复杂就是“先检索再生成”。用户问一个问题先从准备好的文档库里找出最相关的内容片段把这些片段拼进提示词里再丢给大模型让它基于这些材料作答。这个思路听着简单真做起来就会发现事情没那么轻松。文档格式五花八门有 PDF、Word、Markdown、HTML光是解析和清洗文本就够写几百行代码。切分策略要反复调试向量数据库的选型和索引参数也有讲究更别提检索结果怎么排序、怎么和生成环节衔接这些问题。我一开始也想自己写毕竟 RAG 的流程不算高深用 requests 库调 API 加上一个向量库客户端也能拼出来。但写着写着就发现自己造轮子的成本远高于预期。光是“文档加载器”这一块LangChain 已经封装了上百种格式的解析器PDF 有 PyPDFLoaderWord 有 Docx2txtLoaderMarkdown 有 UnstructuredMarkdownLoader每种都处理了编码、表格、页眉页脚这些烦人的细节。另外像 Prompt 模板、输出解析器、链式调用这些都做得非常成熟。还有一个很现实的原因——团队协作和后期的维护成本。自己用手写代码拼的流程逻辑分散在各个文件里后来接手的人要看半天才能看懂。LangChain 做的事情就是把这些组件标准化每个环节都是一个可替换的模块新人上手快出了问题排查也方便。说白了LangChain 在这个项目里扮演的角色就是“装配车间”把文档处理、向量检索、模型调用这些零件按顺序组装起来。1.2 为什么必须用向量数据库以及哪款适合你RAG 的核心检索逻辑是“语义相似度检索”也就是把用户的问题变成一组向量然后去向量数据库里找和这组向量最接近的文本片段。传统数据库做不了这件事因为它是基于关键词和精确匹配的你说“怎么申请年假”它找不到文档里写着的“休假管理规定”。所以向量数据库是 RAG 系统里绕不开的一个组件。市面上的选择不少Chroma 适合个人项目跑原型Milvus 适合大数据量和高并发的场景PGVector 适合不想额外维护一个数据库服务的团队因为可以直接利用现有的 PostgreSQL 实例。我在这个项目里用了 Chroma 做开发调试后期迁移到生产环境时换成了 PGVector文章里会把两种方案的部署方式和切换方法都写清楚。先说说我的选型标准。数据量在百万级文档以内QPS 要求不高直接上 Chroma 就够了它的零配置特性让开发效率提升非常明显。但如果是企业级应用需要对接已有的权限体系或者对查询延迟有硬性要求那就得考虑 PGVector 或 Milvus。还有一个容易忽略的点——向量数据库的备份和恢复机制。Chroma 的持久化目录复制到另一台机器上就能用但这个特性同时也是隐患并发写入容易出现数据损坏。PGVector 就没这个问题数据由 PostgreSQL 统一管理事务和备份都很成熟。1.3 国产大模型怎么接API方案和本地部署怎么选国产大模型这几年的API服务已经非常成熟了国内常见的几款——通义千问qwen系列、DeepSeek、智谱清言的GLM系列、Kimimoonshot——都提供了兼容OpenAI格式的接口。这意味着你只需要把base_url换成对应的地址把api_key换成自己的密钥其余代码几乎不用改。选择 API 方案还是本地部署取决于你的业务场景。API 方案的优势是启动快、免运维、按量付费适合中小型项目快速验证本地部署的优势是数据不出内网、隐私性更强、长期使用成本可控但需要准备显卡资源。以通义千问的 qwen-plus 模型为例API 调用延迟在 2 秒左右上下文窗口足够容纳我们的检索结果在知识库问答场景下表现是比较均衡的。如果选择本地部署目前比较主流的方案是 Ollama 加 Qwen 或 GLM 的开源版本。比如 Qwen2.5-7B-Instruct 在消费级显卡上就能跑得动量化后显存占用控制在 8GB 左右我用一台 3060 12G 的机器实测生成速度大概在每秒 15 到 20 个 token做内部知识库问答够用了。DeepSeek 的开源模型比如 DeepSeek-V3性能更强但对显存的要求也高了不少一般是团队有 A100/H100 级别资源才会考虑。2. 环境准备与依赖安装2.1 Python版本与虚拟环境避坑指南这个项目我用了 Python 3.10这个版本兼容性最稳。Python 3.11 和 3.12 也能跑但有些库比如 onnxruntime 的某些版本在 Windows 上可能找不到预编译的 wheel 包装起来比较折腾。虚拟环境一定要建别图省事直接往全局环境里装。我见过太多人把 langchain、torch、numpy 这些包直接 pip install 到系统 Python 里装到一半发现和已有项目的依赖冲突要么卸载重装要么破罐子破摔。用 venv 和 conda 都可以我个人习惯用 conda因为 Python 版本切换比较方便。提示在 Windows 上创建 conda 环境的时候指定 Python 3.10 这个版本最稳妥。2.2 安装核心依赖包清单及版本说明核心依赖就这几个langchain框架主体langchain-community包含了各种文档加载器和向量库适配器langchain-openai用来对接兼容 OpenAI 格式的国产大模型 APIchromadb本地向量数据库fastapi 和 uvicorn部署 API 服务python-dotenv管理 API 密钥pypdf 和 docx2txt解析 PDF 和 Word 文档用tiktoken计算 token 数量用来控制切分长度安装命令pip install langchain langchain-community langchain-openai chromadb pip install fastapi uvicorn python-dotenv pip install pypdf docx2txt tiktoken这里要说一个坑langchain 这个库从 0.1.0 版本开始做了大规模的模块拆分以前一个包包含所有东西现在变成了 langchain 只有核心的链逻辑文档加载器和各种第三方集成全部分散到 langchain-community 里。如果你直接from langchain.document_loaders import PyPDFLoader在新版本里会报错正确写法是from langchain_community.document_loaders import PyPDFLoader。2.3 获取国产大模型API Key的详细流程以通义千问为例。去阿里云百炼控制台注册账号完成实名认证然后在模型广场找到 qwen-plus 模型点击“开通服务”并创建 API-KEY。这个过程大概十分钟就能搞定新用户一般都有免费额度做实验绰绰有余。DeepSeek 就在开放平台注册账号创建 API Key 就行充值最低 10 块钱够跑上百次问答了。拿到 API Key 之后在项目根目录创建一个.env文件DASHSCOPE_API_KEY你的通义千问key DEEPSEEK_API_KEY你的DeepSeek key然后写一个工具函数来加载环境变量from dotenv import load_dotenv load_dotenv()需要说明的是LangChain 对接国产大模型有两种方式。如果模型厂商提供了兼容 OpenAI 的接口直接用ChatOpenAI指定base_url就行如果厂商自己的 SDK 封装得比较好也可以继承 LangChain 的BaseChatModel类做定制封装。我后面代码里用的是第一种方式简单直接。2.4 Embedding模型的选用与部署细节Embedding 模型这一步很多人会忽略。其实 RAG 系统的回答质量瓶颈大多数时候不在大模型而在检索环节。检索的效果取决于两件事切分策略合不合理以及 Embedding 模型的语义理解能力。国内使用比较友好的是阿里云的 text-embedding-v3 系列通过 API 调用效果不错。如果不想用 API有个偏轻量的开源方案是 BAAI/bge-small-zh-v1.5这个模型只有 24M 参数CPU 上就能跑对中文的支持和维度控制都做得比较好。我项目里用的是 API 方案调用方式from langchain_openai import OpenAIEmbeddings embeddings OpenAIEmbeddings( modeltext-embedding-v3, base_urlhttps://dashscope.aliyuncs.com/compatible-mode/v1, api_keyos.getenv(DASHSCOPE_API_KEY) )Embedding 的维度也值得关注。text-embedding-v3 默认是 1024 维bge-small-zh-v1.5 是 512 维。维度越高代表每个文本的语义特征越丰富检索的精细度也会更好但相应的存储空间和计算耗时都会增加。在数据量不是特别大的场景下512 维和 1024 维的差距并不明显。选模型的时候还是优先看语义匹配的效果单看数字意义不大。3. 核心实现从文档加载到问答输出3.1 文档加载与解析的完整处理流程文档加载是整个 RAG 流程的入口也最容易踩坑。PDF 文件的解析是个老大难问题扫描版的 PDF 需要 OCR文字版的 PDF 还会遇到表格错乱、页眉页脚混入正文的情况。我实测下来PyPDFLoader 对文字版 PDF 的解析效果尚可但遇到分栏排版或者表格内容较多的文档结构会丢失得比较厉害。相关代码from langchain_community.document_loaders import PyPDFLoader from langchain_community.document_loaders import Docx2txtLoader from langchain_community.document_loaders import TextLoader from langchain.text_splitter import RecursiveCharacterTextSplitter def load_document(file_path): if file_path.endswith(.pdf): loader PyPDFLoader(file_path) elif file_path.endswith(.docx): loader Docx2txtLoader(file_path) else: loader TextLoader(file_path, encodingutf-8) return loader.load()一个实际经验很多企业内部知识库的文档都是 PDF 格式但 PDF 的排版没有统一标准有的带目录、有的带水印、有的页脚是公司名。如果你在加载之后直接做切分页脚的水印文本会混进知识库检索的时候动不动就把这些无关内容匹配出来。所以文档清洗这一步很重要我一般在加载之后会做一次正则过滤把明显无关的内容剔除掉。注意不要指望 LangChain 自带的 loader 能完美处理所有 PDF。如果实测下来解析效果不理想建议先用 pypdf 或者 pdfplumber 做预处理把内容按“阅读顺序”重新整理好再交给 LangChain。我遇到分栏的 PDF 时会用 pdfplumber 手动提取坐标和文本然后按坐标顺序重排虽然代码多写了三五十行但检索准确率的提升是肉眼可见的。3.2 文本切分策略的核心参数讲解文本切分是 RAG 里最被低估的环节。切分的粒度直接决定了检索的“命中精度”。如果切分得太粗比如把一个 50 页的文档整个丢进向量库那检索出来的是一个超大文档里面相关内容只占很小一部分大模型读了大部分无关内容回答的质量自然就差了。如果切分得太细比如按一句话一切那检索出来的片段缺少上下文大模型看到的信息是孤立的也没办法给出准确回答。我实测了近十种切分策略最稳定的是 RecursiveCharacterTextSplitter。它的原理是递归地按照一系列分隔符来切分文本优先保留语义完整的段落和句子。参数设置如下from langchain.text_splitter import RecursiveCharacterTextSplitter text_splitter RecursiveCharacterTextSplitter( chunk_size500, chunk_overlap50, separators[\n\n, \n, 。, , , ., !, ?, , ;, , ,, , ] )chunk_size 是每个片段的长度上限按字符数计算chunk_overlap 是相邻片段之间的重叠长度。两个参数的核心逻辑是既要保证单个片段有足够的上下文信息又不能超出 Embedding 模型的最大输入长度和检索匹配的精度范围。选择 500 这个值是因为我测过大概 300 到 800 这个区间300 时片段太碎检索出的内容经常答非所问800 时片段太长容易混入过多无关信息500 是一个比较平衡的点。chunk_overlap 设为 50 是为了避免一个完整句子被从中间截断让前后两个片段各保留一点交叉信息这样即使需要跨片段的信息检索也能命中。还有一个容易被忽略的问题chunk_size 的计量单位。LangChain 的 RecursiveCharacterTextSplitter 默认按字符数计算对中文来说这个方式基本够用。但如果你的文档是中英混排建议换用 TokenTextSplitter按 token 数来切分这样对模型的提示词长度控制更精确。切分完成之后还有一个可选但强烈建议的步骤——用 metadata 标记每个片段的来源信息包括文件名、页码、章节标题。这样检索到结果后系统能告诉用户“这个答案来自《员工手册》第 3 页”可信度和使用体验会好很多。3.3 向量化与知识库入库的实现方式把所有切分好的文本片段交给 Embedding 模型生成向量然后写入向量数据库。这个过程在 LangChain 里一行代码就能完成from langchain_community.vectorstores import Chroma vectorstore Chroma.from_documents( documentsdocs, embeddingembeddings, persist_directory./data/chroma_db )这里有个性能问题值得聊一聊。Chroma.from_documents是逐条向量化再写入的如果文档量很大比如几千个 PDF这个过程会非常慢。我做过一个 2000 个 PDF 的导入实验用默认方式跑了一晚上才完成。后来改用批处理方式你自己把文本分批传给 Embedding 接口然后把生成的向量用Chroma.add_embeddings批量写入时间直接缩短到原来的三分之一。核心是 Embedding API 的并发请求限制不同的服务商对 QPS 的限制不同实际使用的时候要做一层并发控制既不能太慢也不要把 API 打挂。入库之后别忘了做一次简单的验证。从库里随机抽几个问题用相似度检索看能不能返回相关的片段。这一步能帮你判断前面的切分和向量化效果不要等整个系统搭完了才发现知识库压根检索不到正确内容。3.4 检索增强与问答链实现代码详解系统走到这一步开始做真正的“检索增强”。LangChain 提供了一个叫 RetrievalQA 的链把所有环节串起来from langchain.chains import RetrievalQA from langchain_openai import ChatOpenAI llm ChatOpenAI( modelqwen-plus, base_urlhttps://dashscope.aliyuncs.com/compatible-mode/v1, api_keyos.getenv(DASHSCOPE_API_KEY), temperature0.3 ) qa_chain RetrievalQA.from_chain_type( llmllm, chain_typestuff, retrievervectorstore.as_retriever(search_kwargs{k: 4}), return_source_documentsTrue )这段代码里最值得斟酌的是chain_typestuff和search_kwargs{k: 4}这两个参数。stuff 模式是把检索到的所有片段直接拼接到提示词里适合片段数量少一般不超过 5 个且总长度在模型上下文窗口内的场景。如果你的知识库结构复杂、片段数量多可以考虑 map_reduce 或 refine 模式。map_reduce 会对每个片段单独提问再汇总答案效果更稳定但调用成本翻倍。refine 则是在已有答案的基础上逐步修正适合迭代优化场景。实际项目里我建议先从 stuff 开始如果回答质量不达标再上 refine。k4表示返回最相似的 4 个片段。这个值不是越大越好片段太多会造成两个问题一是提示词过长模型可能混淆重点二是无关信息可能被噪声干扰反而降低了回答的准确性。我试过 k 从 2 到 8 的区间4 到 5 是效果和成本的平衡点。再来看 Prompt 的设计。默认的 Prompt 是英文模板直接用在中文场景会出现表达风格漂移模型会用英文答辩式的口吻回答中文问题。所以我在实际项目中自定义了中文 Promptfrom langchain_core.prompts import PromptTemplate prompt_template 你是一个知识库问答助手请根据以下已知信息用中文简洁、准确回答用户的问题。 已知信息 {context} 用户问题{question} 回答要求 1. 如果已知信息中有相关内容请基于这些内容进行回答并注明内容来源。 2. 如果已知信息中没有相关内容请直接回答“知识库中没有找到相关内容”禁止编造。 3. 回答要条理清晰重点突出。 回答 QA_CHAIN_PROMPT PromptTemplate( templateprompt_template, input_variables[context, question] ) qa_chain RetrievalQA.from_chain_type( llmllm, chain_typestuff, retrievervectorstore.as_retriever(search_kwargs{k: 4}), return_source_documentsTrue, chain_type_kwargs{prompt: QA_CHAIN_PROMPT} )这个 Prompt 模板里面“禁止编造”这条约束在 RAG 系统里非常关键。一个常见的误区是完全信任大模型不做任何约束。实际上大模型在知识库没有相关内容时很容易“一本正经地胡说八道”把这个禁止条款写进提示词能显著减少这种幻觉。3.5 完整代码串联与使用效果展示把所有环节组装起来完整的问答函数也就几十行代码import os from dotenv import load_dotenv from langchain_community.document_loaders import PyPDFLoader, Docx2txtLoader from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain_openai import ChatOpenAI, OpenAIEmbeddings from langchain_community.vectorstores import Chroma from langchain.chains import RetrievalQA from langchain_core.prompts import PromptTemplate load_dotenv() embeddings OpenAIEmbeddings( modeltext-embedding-v3, base_urlhttps://dashscope.aliyuncs.com/compatible-mode/v1, api_keyos.getenv(DASHSCOPE_API_KEY) ) CHROMA_PATH ./data/chroma_db def build_knowledge_base(file_paths): all_docs [] for fp in file_paths: if fp.endswith(.pdf): loader PyPDFLoader(fp) elif fp.endswith(.docx): loader Docx2txtLoader(fp) else: continue all_docs.extend(loader.load()) splitter RecursiveCharacterTextSplitter( chunk_size500, chunk_overlap50, separators[\n\n, \n, 。, , , ., !, ?, , ;, , ,, , ] ) docs splitter.split_documents(all_docs) vectorstore Chroma.from_documents( documentsdocs, embeddingembeddings, persist_directoryCHROMA_PATH ) return vectorstore def create_qa_chain(vectorstore): llm ChatOpenAI( modelqwen-plus, base_urlhttps://dashscope.aliyuncs.com/compatible-mode/v1, api_keyos.getenv(DASHSCOPE_API_KEY), temperature0.3 ) prompt_template 你是一个知识库问答助手请根据以下已知信息用中文简洁、准确回答用户的问题。 已知信息 {context} 用户问题{question} 回答要求 1. 如果已知信息中有相关内容请基于这些内容进行回答并注明内容来源。 2. 如果已知信息中没有相关内容请直接回答“知识库中没有找到相关内容”禁止编造。 3. 回答要条理清晰重点突出。 回答 QA_CHAIN_PROMPT PromptTemplate( templateprompt_template, input_variables[context, question] ) return RetrievalQA.from_chain_type( llmllm, chain_typestuff, retrievervectorstore.as_retriever(search_kwargs{k: 4}), return_source_documentsTrue, chain_type_kwargs{prompt: QA_CHAIN_PROMPT} ) if __name__ __main__: vs build_knowledge_base([./data/handbook.pdf]) chain create_qa_chain(vs) result chain.invoke({query: 年假可以分几次休}) print(回答:, result[result]) print(来源:, [doc.metadata.get(source) for doc in result[source_documents]])这版代码在本地跑通之后我用一份 50 页的员工手册做了测试问了几个问题比如“带薪年假的享受条件是什么”“薪资发放日期是哪天”回答都能从文档中找到对应内容而且标注了来源页数。当然也遇到过答非所问的情况比如把产品说明书里的术语和员工手册里的内容混淆了这就要靠后面的混合检索和重排序来优化。4. 用FastAPI把RAG服务暴露成HTTP接口4.1 项目目录结构与FastAPI应用搭建前面写的代码在命令行里跑通了但知识库问答系统的价值要发挥出来必须提供一个 API 服务让前端页面、企业微信机器人、钉钉应用等都能调用。我把项目拆成了模块化的结构rag_project/ ├── .env ├── requirements.txt ├── app.py # FastAPI 入口 ├── rag/ │ ├── __init__.py │ ├── loader.py # 文档加载 │ ├── splitter.py # 文本切分 │ ├── vectorstore.py # 向量库操作 │ ├── qa_chain.py # 问答链 │ └── config.py # 全局配置 ├── data/ │ └── kb_docs/ # 知识库原始文档 └── vectorstore/ └── chroma_db/ # 向量数据库持久化目录FastAPI 这个框架接口很直观一个基础服务大概长这样from fastapi import FastAPI, UploadFile, File, HTTPException from pydantic import BaseModel app FastAPI(titleRAG 知识库问答系统) class QueryRequest(BaseModel): question: str top_k: int 4 class QueryResponse(BaseModel): answer: str sources: list[str] app.post(/qa, response_modelQueryResponse) async def answer_question(req: QueryRequest): try: result qa_chain.invoke({query: req.question}) return QueryResponse( answerresult[result], sources[doc.metadata.get(source, ) for doc in result[source_documents]] ) except Exception as e: raise HTTPException(status_code500, detailstr(e)) app.post(/upload) async def upload_document(file: UploadFile File(...)): # 保存文件并更新知识库 ...4.2 接口设计精髓与文档上传功能实现真正上生产接口设计要考虑的细节远比上面这个 demo 多。/qa接口做问答/upload接口做文档上传这两个接口的职责要分开因为文档上传后的向量化入库是异步耗时的操作不能阻塞问答接口的响应。我的做法是/upload接口接收文件后先保存到磁盘返回一个document_id把“文档解析 - 切分 - 向量化 - 入库”这个流程丢到后台任务队列里执行/qa接口查询的时候只从已经完成入库的向量库里检索文档上传功能的实现import uuid from pathlib import Path from fastapi import BackgroundTasks UPLOAD_DIR Path(./data/kb_docs) UPLOAD_DIR.mkdir(exist_okTrue) app.post(/upload) async def upload_document(file: UploadFile File(...), background_tasks: BackgroundTasks None): doc_id str(uuid.uuid4()) suffix Path(file.filename).suffix save_path UPLOAD_DIR / f{doc_id}{suffix} with open(save_path, wb) as f: content await file.read() f.write(content) background_tasks.add_task(process_document, str(save_path), doc_id) return {document_id: doc_id, status: processing} def process_document(file_path: str, doc_id: str): try: docs load_document(file_path) splitter RecursiveCharacterTextSplitter( chunk_size500, chunk_overlap50, separators[\n\n, \n, 。, , , ., !, ?, , ;, , ,, , ] ) chunks splitter.split_documents(docs) for chunk in chunks: chunk.metadata[doc_id] doc_id vectorstore get_vectorstore() vectorstore.add_documents(chunks) vectorstore.persist() except Exception as e: print(f处理文档失败: {e})因为这个接口在同一时间可能被多个用户调用向量数据库的并发写入需要特别小心。Chroma 在并发写入时容易出问题如果服务端有多个 worker建议在写入前加一个线程锁或者直接换成 PGVector。4.3 登录鉴权与请求日志生产环境的几个关键点这一步属于生产环境的基本要求了没有鉴权的 API 相当于把知识库完全开放哪怕只是内网使用风险也很大。鉴权方案我用的是最简单有效的 API Key 机制。在.env里配置一个API_AUTH_TOKEN然后在 FastAPI 里加一个依赖函数from fastapi import Depends, Header API_AUTH_TOKEN os.getenv(API_AUTH_TOKEN, secret-token) async def verify_token(authorization: str Header(...)): if authorization ! fBearer {API_AUTH_TOKEN}: raise HTTPException(status_code401, detail认证失败) app.post(/qa, response_modelQueryResponse, dependencies[Depends(verify_token)]) async def answer_question(req: QueryRequest): ...请求日志这一块也别省。每次问答请求记下问题内容、回答内容、检索到的来源文档、响应时长和状态码。日志累积之后可以用来分析用户的提问分布比如哪些问题是知识库没覆盖到的哪些文档经常被检索到这些都是优化知识库的宝贵数据。4.4 启动服务的正确姿势与调试技巧启动 FastAPI 服务官方推荐用 uvicornuvicorn app:app --host 0.0.0.0 --port 8000开发调试的时候记得加--reloaduvicorn app:app --host 0.0.0.0 --port 8000 --reload生产环境部署建议加几个参数--workers 4配合 Gunicorn 做多进程管理--timeout 120防止长请求超时被误杀。我一个实际经验是大模型生成速度如果不是流式接口有可能超过默认的 30 秒超时设置。要么把超时设大要么改造成流式输出。流式输出对用户体验的提升非常明显建议优先做这件事。注意多进程模式下Chroma 持久化目录的并发访问有损坏风险。稳妥方案是把知识库存储切换到 PostgreSQLPGVector或者用 Redis 做一层分布式锁。如果数据量不大直接单进程部署也可以接受但要有明确的容量规划。5. 生产级RAG系统优化检索质量与性能调优5.1 混合检索策略向量检索与关键词检索的融合纯向量检索有个明显短板——对专有名词、产品型号、员工编号这类“精确信息”的支持不够好。比如用户问“型号 ABC-100 的功率是多少”向量检索可能把它语义扩展到“设备参数”“运行效率”反而忽略了最关键的型号字符串。这时候就需要引入关键词检索来做精确匹配。我在项目里实现了一套混合检索逻辑向量检索负责语义召回BM25 负责精确关键词匹配两者取并集再通过分数融合排序。LangChain 提供了EnsembleRetriever来组合各种检索器from langchain.retrievers import BM25Retriever, EnsembleRetriever from langchain_community.vectorstores import Chroma vector_retriever vectorstore.as_retriever(search_kwargs{k: 4}) bm25_retriever BM25Retriever.from_documents(chunks, k3) ensemble_retriever EnsembleRetriever( retrievers[bm25_retriever, vector_retriever], weights[0.4, 0.6] ) qa_chain RetrievalQA.from_chain_type( llmllm, chain_typestuff, retrieverensemble_retriever, return_source_documentsTrue, chain_type_kwargs{prompt: QA_CHAIN_PROMPT} )权重系数 0.4 和 0.6 是我实测后定的。BM25 权重太高会让回答变得“死板”只按字面匹配向量检索权重太高又丢掉了精确匹配的优势。这个比例在不同领域的知识库上表现有差异工程实践中需要根据自己的数据做小规模的调参测试。5.2 重排序Rerank机制提升答案准确率的关键一步初检阶段拿到的片段往往有噪声比如某个片段同时包含了好几个主题的内容而用户问的只是其中一个。这时候如果把这些混杂的片段直接拼给大模型回答的准确性会受影响。解决方案是加一个重排序层用一个更强的模型对初检结果进行精细打分把最相关的片段排到最前面。我用的方式是调用一个独立的 Rerank API 或者本地部署一个 rerank 模型比如 bge-reranker。处理逻辑是初检拿到 10 个候选片段Reranker 逐条计算与问题的相关度分数最后只取分数最高的 4 个喂给大模型。这一层“粗排 精排”的思路在搜索系统里很成熟用到 RAG 里同样效果显著。我实测过加了 Rerank 之后回答准确率大概提升了 15% 到 20%尤其是在知识库内容接近、模糊问题较多的场景下效果提升很明显。代价是查询响应时间增加了大概 300 到 500 毫秒但换来的准确率提升完全值得。5.3 缓存机制用语义缓存降低重复请求开销知识库问答场景有一个特点很多问题会被反复提问。比如“怎么修改密码”这个问题几十个员工会在一周内反复问。每次都对相同问题做完整检索和生成浪费计算资源不说响应时间也不稳定。解决方案是做一层语义缓存。我的实现思路是把用户的问题向量化在缓存区里做相似度匹配如果找到相似度超过 0.95 的历史问题就直接返回历史答案不再走大模型。这个方案是语义级别的缓存和传统对输入文本做精确匹配的缓存不同。class SemanticCache: def __init__(self, max_len1000): self.cache [] self.max_len max_len def get(self, question_vector): if not self.cache: return None # 简单循环比对生产环境建议用向量数据库实现 for cached_qv, answer in self.cache: similarity cosine_similarity(question_vector, cached_qv) if similarity 0.95: return answer return None def put(self, question_vector, answer): self.cache.append((question_vector, answer)) if len(self.cache) self.max_len: self.cache.pop(0)实测效果非常可观。知识库问答系统上线后前两周的缓存命中率大约在 30% 到 40%基础问题的响应时间直接降到 100 毫秒以内。这里要注意缓存的过期策略——知识库更新之后旧缓存需要主动失效否则用户会拿到过时信息。5.4 提示词压缩技术让回答更精准不跑偏检索到的片段里真正和问题强相关的内容往往只占一部分。有时候我们把 4 个片段拼进提示词其中 2 个完全不相关这会让大模型在生成回答时被干扰。解决这个问题除了 Rerank还可以做提示词压缩。LangChain 有个ContextualCompressionRetriever的机制。它让一个大模型或者 reranker 先看片段内容和用户问题把片段中与问题不相关的部分过滤掉只保留最核心的文本再进入最终提示词。这样做的好处是大幅减少了提示词中的噪声回答时模型可以更聚焦。from langchain.retrievers import ContextualCompressionRetriever compression_retriever ContextualCompressionRetriever( base_compressorreranker_model, base_retrieverensemble_retriever )理论上这层会再次增加延迟因为每个片段都要过一遍压缩模型。但在追求高质量回答的场景中这层是最有效的“提纯”手段。我一般是按需开启如果知识库内容噪声大、检索结果经常跑偏就开启如果知识库本身质量高、用户问答也比较集中这层可以不开。6. 常见问题与排查技巧实录6.1 高频报错整理与解决方案速查表调试过程中遇到最多的问题我用一个表格整理出来方便大家对照排查报错信息原因分析解决方案ModuleNotFoundError: No module named langchain_community安装的 LangChain 版本过新文档加载器已拆分为独立包pip install langchain-community改从langchain_community导入OpenAIError: Invalid base_urlbase_url 拼写有误或者漏掉了兼容模式路径检查是否写成https://dashscope.aliyuncs.com/compatible-mode/v1完整地址RateLimitError调用国产大模型 API 频率过高加入指数退避重试机制或者调整并发数chromadb.errors.UniqueConstraintError重复添加了相同 ID 的文档片段写入前检查doc_id是否冲突使用uuid4确保唯一性FastAPI 接口超时同步调用大模型生成耗时长于默认超时设置改为流式响应或调大 uvicorn 的timeout检索不到任何内容向量库为空或 Embedding 维度不匹配检查向量库是否有数据确认embedding模型是否一致回答内容与知识库无关chunk_size 过大或过小导致语义不聚焦调整切分参数为 500/50增加混合检索和 RerankPyPDFLoader 解析乱码PDF 是扫描版或使用了特殊编码改用 OCR 工具预处理或使用 pdfplumber 提取6.2 踩坑实录几个让人印象深刻的调试经历第一个坑是 Embedding 模型不一致导致的“检索不到内容”。我在开发机用 bge-small-zh-v1.5 生成了向量并存入 Chroma然后部署到服务器上时换了 text-embedding-v3结果所有查询都返回空结果。排查了很久最后发现两个模型生成的向量维度不一致512 维和 1024 维相似度计算全部失效。这个问题的教训是从开发到生产的整个链路里Embedding 模型必须一致最好在配置文件中写死不能靠人工记忆。第二个坑是 PDF 表格解析的失真。我用 PyPDFLoader 加载一份带表格的技术参数文档切分后的文本完全乱掉了表格的列标题和数据行被拆分到不同片段检索出来给大模型一解读就全错了。后来我改用 pdfplumber 专门处理这类表格先提取表格结构再按行转成文本效果好了很多。遇到大段文字加表格混排的文档时我会单独写一个预处理函数先按坐标和类型段落、表格、页眉页脚分区域提取再按阅读顺序重组。第三个坑是 FastAPI 多进程模式下 Chroma 的锁冲突。我在生产环境用 Gunicorn 起了 4 个 worker结果上传文档和问答同时进行时Chroma 经常报database or disk is full或者直接卡死。查了一圈发现是 SQLite 后端的并发写入问题。后来我把向量库换成了 PGVector问题就彻底解决了。如果数据量和 QPS 不大也可以用单 worker 线程锁的方式应对。6.3 新模型库跑通后的一些优化心得整套系统跑通之后我觉得从“能用”到“好用”还有三件事值得做。第一件事是知识库的更新机制。企业里的文档不是一成不变的制度会更新产品说明会修订。RAG 系统要有一个清晰的知识版本管理方案——上传新文档时旧版本的向量是保留还是删除文档内容变了之前缓存的答案要不要清理我的做法是给每个片段打上文档版本号更新时按doc_id删除旧向量再写入新向量缓存按周期自动失效。第二件事是回复的可解释性。在给用户的回答中标注来源文档和具体页码这个体验在内部系统中尤其重要。员工看到答案来自哪份文件信任度会高很多。实现上已经做了return_source_documentsTrue就能拿到来源关键是把这些信息展示到前端界面上。第三件事是持续的评测与优化。我建了一个包含 200 条测试问题的评估集每次调整切分参数、检索策略、Prompt 模板之后都要跑一遍回归测试对比回答准确率的变化。没有这个评估集所有优化都是“凭感觉”很难判断某个改动到底改善了还是劣化了。这个过程虽然枯燥但它是让知识库问答系统走向生产级的必经之路。最后再聊一点个人体会。做这类系统别一上来就追求“大而全”。先用最简单的方式把问答链路跑通用一个小文档库验证效果然后一个模块一个模块地往里加优化——先加混合检索再加 Rerank再上缓存每一步都做对比评测。这样迭代方向清晰出现问题时也能快速定位。我见过太多人一开始就把 Milvus、LangGraph、Agent 这些概念堆上去结果系统复杂到出了问题根本不知道往哪里查。技术栈再先进能解决问题、稳定上线才是硬道理。