很多人第一次做RAG知识库问答,最容易卡住的地方不是“不知道RAG是什么”,而是“代码怎么落地”。文档怎么加载、文本怎么切、向量怎么存、检索怎么接、Prompt怎么拼,每一步翻文档都能找到零散答案,但串起来总在报错。这篇内容我直接把一套用LangChain配合通义千问实现的可运行RAG知识库问答代码拆开讲清楚,重点解释每个环节为什么这么做,以及实际运行中会踩到哪些坑。整套代码不需要自己搭服务,跑通后你就能拿自己的本地文本文件做召回式问答,后续接PDF、Word、增量更新也都有明确思路。
1. 项目概述:RAG知识库要解决什么核心问题
1.1 没有检索增强时,大模型问答的尴尬点
先说一个很常见的现象。有人把几十页产品手册直接丢给大模型,然后问“保修政策第三条具体怎么执行”,模型会给出一段看起来合理但实际细节对不上的回复。这不是模型笨,而是它根本没有“看过”你那本手册。大模型的训练知识来自公开语料,企业内部文档、新版本操作说明、产品迭代后的新规则,模型一概不知道,强行回答就是一本正经地编。
RAG(检索增强生成)的思路,是在把问题交给模型之前,先从你的知识库里把最相关的片段捞出来,然后把这些片段作为参考资料一起喂给模型。模型回答的依据来自你提供的材料,而不是它记忆里的模糊印象,准确率、可控性都会好很多。我自己的体会是:对于“企业内部知识问答”这类场景,RAG带来的提升要远大于换一个更大的模型。
1.2 为什么选LangChain加通义千问这套组合
做RAG方案,市面上可选的组合很多。LangChain属于流程编排层,它把文档加载、文本切分、向量化存储、检索器、Prompt模板、大模型调用这些环节全部抽象成统一接口,代码写起来像搭积木,后期想替换某个组件也很方便。通义千问作为大模型服务,中文理解和长文本表现都够用,接入方式是标准的API调用,不需要自己去部署模型,对个人开发者和中小团队来说是最省事的一条路径。
我做过几个不同项目,框架层面换过别的编排工具,最后又转回LangChain,原因就是社区资料多、踩坑经验容易搜到,而且它的向量库、检索器、文档加载器等模块基本覆盖了所有常见需求。哪怕你以后想换成其他国产模型或者开源模型,整体代码结构保持不动,只需要替换模型接口,这是LangChain架构比较舒服的地方。
1.3 跑通它能得到什么,适合哪些人
这套代码最终会给你一个命令行交互程序。你把自己的文本文件放进指定目录,程序会对文档内容进行切分、向量化、构建索引,然后你可以直接在终端里问问题,它会结合知识库内容给出答案,并返回答案引用的文档片段。
适合的人群有两类:一类是刚开始接触RAG的开发者,可以把它当作练手项目,把流程跑通后再逐步加功能;另一类是工作中确实有内部知识库问答需求但时间紧张的人,代码改一改文档路径和提问格式,就能快速出一个可用原型。
2. 环境准备:依赖、密钥与测试文档
2.1 依赖安装与版本避坑说明
建议使用Python 3.9以上版本,我实测在3.10和3.11环境都正常。需要安装的核心依赖如下:
pip install langchain langchain-community langchain-text-splitters pip install dashscope pip install faiss-cpu如果后续要支持PDF,还需要安装:
pip install pypdf这里提醒一下:LangChain从较新的版本开始,很多模型接入类和向量库类被拆分到了langchain-community包里,所以导入路径跟老教程有些不一样。比如通义千问的对话模型在langchain_community.chat_models.tongyi里,Embedding模型在langchain_community.embeddings.dashscope里。如果你用的是老版本,导入路径可能是langchain.llms.tongyi这种,两种写法对应不同版本,千万别混着用。
2.2 DashScope密钥配置与模型选择
去阿里云百炼或DashScope平台开通对应的服务,然后创建API Key,配置到环境变量里:
export DASHSCOPE_API_KEY="你的密钥"Windows下可以用:
set DASHSCOPE_API_KEY=你的密钥不建议把密钥硬编码写在代码里,尤其是代码要发布到公共平台的时候,见过太多人因为把Key提交到仓库里导致被刷余额,千万别再踩这个坑。
模型选择方面,问答生成可以选择qwen-plus或qwen-turbo,qwen-plus综合能力更好,qwen-turbo更快更省。Embedding向量化选择text-embedding-v2或text-embedding-v3都可以。如果你的账号是新开的,建议先确认一下目标模型是否已经开通权限,否则调用时会报InvalidApiKey或Model not exist这类错误。
2.3 准备一份合适的测试文档
测试文档最好是一份结构清晰的Markdown或TXT文件,内容在几千字以上,包含明确的章节、步骤说明、常见问题等自然分段信息。不要拿只有几十行的高重复文本去测,那种测试结果容易误导你。
我自己的经验是:先用一份“员工入职操作手册”这类内容测,因为它天然分成“入职流程”、“账号申请”、“系统权限”、“常见问题”等段落,方便验证检索效果。测试文档保存为UTF-8编码,避免后续加载时出现乱码。
3. 核心流程拆解与关键代码实现
3.1 文档加载:让系统读入你的私有资料
LangChain的文档加载器负责把不同来源的文件转成标准的Document对象。Document对象包含两个东西:page_content文本内容和metadata元数据信息。
加载单个文本文件的代码如下:
from langchain_community.document_loaders import TextLoader loader = TextLoader("knowledge_docs/knowledge.txt", encoding="utf-8") docs = loader.load()加载多个文本文件可以用DirectoryLoader:
from langchain_community.document_loaders import DirectoryLoader, TextLoader loader = DirectoryLoader( "knowledge_docs/", glob="*.txt", loader_cls=TextLoader, loader_kwargs={"encoding": "utf-8"} ) docs = loader.load()知识库目录 knowledge_docs/ ├── 产品手册.txt ├── 常见问题.txt └── 操作指南.txt调试点:如果文档加载后打印出来发现内容包含一堆不相关的空行、表格符号或者控制字符,建议先做个简单的预处理清洗。因为文档内容质量直接影响切分和向量化的效果,这一步不能图省事。
3.2 文本切分:chunk_size和overlap怎么定
加载进来的文档通常很长,不能直接做向量化,因为单条向量不可能承载太多语义信息,而且超出模型上下文的部分也没意义。所以要把文档切分成小块,这个动作叫chunking。
我推荐用RecursiveCharacterTextSplitter,它比普通的CharacterTextSplitter更聪明,会按照段落、换行、句号、分号等优先级递归切分,尽量保留语义完整性。
from langchain_text_splitters import RecursiveCharacterTextSplitter text_splitter = RecursiveCharacterTextSplitter( chunk_size=500, chunk_overlap=100, separators=["\n\n", "\n", "。", ";", " ", ""] ) chunks = text_splitter.split_documents(docs)chunk_size和chunk_overlap怎么定?我给一个务实的经验值:
- 中文场景,
chunk_size建议300到800之间。 chunk_overlap一般是chunk_size的10%到30%。- 如果文档本身按章节分得很清楚,优先用每章一个完整段落作为chunk,避免强行拆散上下文。
道理其实很简单:chunk太长,检索时噪声多,而且放进Prompt会占用上下文空间;chunk太短,语义不完整,模型可能理解不了上下文。比如一个“退款流程”在上一段,而“两天内到账”被切到了下一段,模型就会丢失关键信息。设置overlap的目的就是要让相邻块之间有重叠区,避免关键信息刚好卡在切断处。
我这里补充一个计算概念:一份2000字的文档,设chunk_size=500、chunk_overlap=100,那它会切成大约5块而不是4块,因为每切到500字后要往回重叠100字再继续切。所以不是简单用总字数除以块大小。
3.3 向量化与向量库:Embedding模型和FAISS
文本切好后,下一道工序是把每个chunk转换成向量表示。这一步用到的Embedding模型会把“文本的语义”压缩成一组固定维度的数字,语义相近的文本,向量距离也会相近。这个概念可以类比成给每个文本片段做“语义指纹”。
代码如下:
from langchain_community.embeddings.dashscope import DashScopeEmbeddings embedding = DashScopeEmbeddings(model="text-embedding-v2")向量库选择上,我建议先用FAISS。FAISS是Meta出的向量检索库,本地运行够快,还能把索引保存到磁盘,不需要单独部署服务端,对入门和中小规模文档非常友好。虽然Chroma、Milvus、Elasticsearch各有优势,但对一个单机知识库问答项目来说,FAISS的“零运维”优势是最直接的。
from langchain_community.vectorstores import FAISS # 从切分好的文档构建向量库 vector_store = FAISS.from_documents(chunks, embedding) # 保存索引到本地 vector_store.save_local("faiss_index") # 下次启动时直接加载,不用重新向量化 vector_store = FAISS.load_local( "faiss_index", embedding, allow_dangerous_deserialization=True )如果你构建索引的文档数量很大,几千上万个chunk,构建过程可能要等几分钟,这是正常现象。但注意,每次运行都重新构建索引很浪费,所以项目里一般要做一个“索引已存在就加载,不存在才构建”的判断,后面完整代码里会有。
3.4 检索器配置:相似度检索怎么调才准
向量库准备好后,我们通过as_retriever把它包装成检索器,用于后续查询。
retriever = vector_store.as_retriever( search_type="similarity", search_kwargs={"k": 4} )search_type参数,我用得比较多的是similarity和mmr。
similarity:按向量距离排序返回最相似的Top K,简单直接。mmr(最大边际相关性):会在相关性之外兼顾多样性,避免捞回来的几个片段内容高度重复。如果一个问题可能对应多个方面,比如“退货政策中的例外情况有哪些”,MMR的效果通常更好。similarity_score_threshold:带相似度阈值过滤的相似度检索,低于阈值的片段直接丢弃,适合期望严格控制“只从确定相关片段里回答”的场景。
k的取值,我建议2到5之间。取太小,信息不够;取太大,噪声变多,Prompt也会过长。从调试经验来看,k=4是大多数场景比较稳的起步值。
一个非常实用的调试点:如果发现检索回来的片段不对,最直接的办法是打印一下检索器返回的文本内容,确认它到底捞到了什么,再决定是调整切分策略还是调整检索参数。这一步比瞎调模型参数有用得多。
3.5 问答链路组装:Prompt映射与RetrievalQA
检索到相关片段之后,需要把这些片段作为上下文,组合成一个带约束的Prompt,再调用通义千问生成答案。LangChain把这一串流程封装成了RetrievalQA链。
from langchain.chains import RetrievalQA from langchain_community.chat_models.tongyi import ChatTongyi from langchain.prompts import PromptTemplate template = """你是一个严谨的知识库问答助手。请仅根据下面提供的资料回答问题。 如果资料中没有对应的依据,请直接回答“根据当前知识库内容,无法找到答案”,不要编造。 资料: {context} 问题: {question} 请用简洁、准确的语言回答:""" prompt = PromptTemplate( input_variables=["context", "question"], template=template ) llm = ChatTongyi(model="qwen-plus", temperature=0.1) qa_chain = RetrievalQA.from_chain_type( llm=llm, chain_type="stuff", retriever=retriever, prompt=prompt, return_source_documents=True )这里chain_type="stuff"表示把所有检索到的片段直接拼接塞进Prompt。因为我们的chunk数量不多、每个chunk也不长,stuff模式最直接也最快。如果chunk特别多导致Prompt超长,才需要考虑map_reduce或refine这类复杂链。对入门项目来说,stuff够用了。
return_source_documents=True这个参数很关键,它让我们在拿到答案的同时,也能看到答案引用了哪些文档片段。调试阶段务必开启,能一眼看出问题出在“检索没捞对”还是“生成没写好”。
4. 场景适配:从txt到PDF,从离线资料到增量更新
4.1 支持PDF、Word多格式文档的方法
很多人第一步就要处理PDF,毕竟现实中的知识库大多是企业制度文件、产品说明书,PDF和Word占比很高。LangChain对这两种格式都有对应的加载器。
from langchain_community.document_loaders import PyPDFLoader pdf_loader = PyPDFLoader("产品手册.pdf") pdf_docs = pdf_loader.load()Word文档:
from langchain_community.document_loaders import UnstructuredWordDocumentLoader word_loader = UnstructuredWordDocumentLoader("产品介绍.docx") word_docs = word_loader.load()注意UnstructuredWordDocumentLoader底层会调用额外的解析库,安装依赖会重一些。如果只是简单读Word里的文字,也可以先把Word另存为TXT或Markdown再走TextLoader,这条路线虽然麻烦一点但最稳。
如果知识库目录里同时有txt和pdf,需要分别加载后合并:
text_docs = TextLoader("a.txt", encoding="utf-8").load() pdf_docs = PyPDFLoader("b.pdf").load() docs = text_docs + pdf_docs不同加载器返回的Document对象结构是一样的,所以后面切分、向量化、检索的逻辑完全复用,这一点LangChain做得比较省心。
4.2 索引复用与增量更新的处理思路
第一次构建完FAISS索引并保存到本地后,后续如果文档没变,不需要重新切分和向量化,直接加载本地索引即可,能省掉一大段时间。
但知识库一定会变。文档更新了,怎么处理?最简单粗暴的做法是删除索引目录,重新构建全部向量。文档总量不大时,这个方法最可靠。文档量大到重新构建耗时过长时,才需要考虑增量更新,这里提供一条参考思路:
from langchain_community.vectorstores import FAISS from langchain.docstore.in_memory import InMemoryDocstore # 用已有的向量库和新Embedding计算的结果合并 new_chunks = ["新增的知识内容"] new_vectors = embedding.embed_documents(new_chunks) vector_store.add_texts(new_chunks) vector_store.save_local("faiss_index")add_texts方式虽然方便,但要注意:FAISS索引不支持删除单条记录,如果文档需要“撤回”或“替换”,分裂状态会很麻烦。我的建议是:增量更新只在“只增不改”的场景用,凡是涉及修改,还是全量重建最省心。因为全量重建也就是多等几分钟的事,换来的是结果可控。
4.3 检索效果不如预期时的调整顺序
实际项目里你会发现,答案不准,大部分问题不在模型,而在检索环节。我总结了一个固定的排查顺序,照着走基本能定位:
- 先看检索回来的
source_documents相关不相关。如果不相关,那就是切分策略、Embedding模型或检索参数的问题,跟大模型无关。 - 如果相关但答案还是不对,再检查Prompt模板。有时候模型回答得啰嗦或跑偏,是Prompt没有给它足够的限制。
- 如果相关且Prompt也没问题,但答案很生硬,可以适当调大
temperature到0.2或0.3,答案是会更自然一点。 - 如果知识库内容本身有缺失,那就是知识覆盖问题,再调代码也没用,要补文档。
检索不相关时,优先调k和相似度阈值,其次调chunk_size和overlap,最后考虑换Embedding模型。不要一上来就换大模型,那是在错误的环节使劲。
5. 常见问题排查与避坑实录
5.1 DashScope接入报错怎么办
我自己跑项目时遇到最多的问题都集中在接入阶段。
一种是401认证失败。检查DASHSCOPE_API_KEY环境变量是否真的被读到,可以在代码里直接打印验证:
import os print(os.getenv("DASHSCOPE_API_KEY") is not None)如果打印出来是False,说明环境变量没生效。这时候要检查你是用哪种方式设置的变量,当前终端窗口是否重新打开了。另外,从代码上传到服务器时,环境变量和本地不一致也是常见原因。
另一种是模型名报错,比如InvalidParameter、Model not exist。很大概率是账号没有开通对应模型的权限,或者模型名多打了空格、写错了大小写。Embedding模型text-embedding-v2和对话模型qwen-plus是不同的服务,要分别确认权限。
还有一种是网络请求超时或连接失败。这种要先确认能正常访问大模型API服务,可以先用一个最小调用测试:
from langchain_community.chat_models.tongyi import ChatTongyi llm = ChatTongyi(model="qwen-plus") print(llm.invoke("你好,简单介绍一下你自己"))如果最小调用通过了,再逐步加配置,定位速度快很多。
5.2 FAISS索引加载报错怎么办
FAISS相关的报错也不少。
最经典的是:构建索引时用的是text-embedding-v2,下一次加载索引时不小心把Embedding模型换成了别的,于是报错说向量维度不匹配。这是因为FAISS索引的维度由构建时的Embedding模型决定,换模型必须重建索引。
还有加载本地索引时的新参数问题,新版LangChain对反序列化安全更严格了,需要在load_local时传入allow_dangerous_deserialization=True,否则会直接抛异常。
如果出现faiss模块导入失败,通常是faiss-cpu没装好,重新执行pip install faiss-cpu即可。在有些精简环境里需要先升级pip再装,否则可能装到损坏的包。
5.3 回答质量差的排查路线
回答质量差,先判断是检索问题还是生成问题。
如果回答内容跟你知识库里的某一段文字很像但组织得不好,通常属于生成问题,调整Prompt和temperature能改善。如果回答里的关键信息根本不在你给的知识库里,但模型照样能编出像模像样的内容,那就是Prompt约束不够,没有强调“只能依据资料作答”。
这里分享一个实用技巧:在Prompt里加入否定约束,明确告诉模型“如果资料中没有相关内容,就回答无法找到,不要根据常识或训练知识发挥”。我加了这个约束之后,瞎编率明显下降。同时把temperature调到0.1到0.2之间,让生成更偏向确定性。
5.4 常见问题速查表
| 现象 | 可能原因 | 处理方法 |
|---|---|---|
| 401认证失败 | API Key未生效 | 检查环境变量是否读取,重新设置密钥 |
| 模型不存在 | 模型名写错或未开通权限 | 核对模型名,确认服务权限 |
| 向量维度不匹配 | Embedding模型不一致 | 删除旧索引,用同一模型重建 |
| 索引加载报安全异常 | 新版需要显式授权 | 添加allow_dangerous_deserialization=True |
| 检索结果不相关 | 切分粗或k值太小 | 调整chunk_size、overlap、k值 |
| 回答内容瞎编 | Prompt约束不够 | 增加“只能依据资料回答”的否定约束 |
| 构建索引很慢 | 文档量太大 | 先少量测试,确认效果后再全量构建 |
| 加载PDF乱码 | 扫描版PDF | 需先做OCR识别,普通加载器无法处理 |
6. 附:完整可运行代码
6.1 代码文件与运行方式说明
下面这份代码我整理成了一个单文件脚本,保存为rag_kb_qa.py即可运行。运行前需要:
- 安装好前面的依赖包
- 设置好
DASHSCOPE_API_KEY环境变量 - 在项目目录下创建
knowledge_docs文件夹,里面放一份knowledge.txt测试文档
启动方式:
python rag_kb_qa.py脚本启动后,如果本地没有索引,会自动加载文档、切分、向量化、保存索引;如果已经有索引,会直接加载,然后进入交互模式。输入问题回车获取答案,输入exit或quit退出。
注意,这里的代码路径和参数都是方便直接复制的。如果你要换成自己的文档,改DOC_PATH、INDEX_DIR、chunk_size这些常量就行。
6.2 完整代码实现
""" RAG知识库问答:LangChain + 通义千问 运行:python rag_kb_qa.py """ import os from langchain_community.document_loaders import TextLoader from langchain_text_splitters import RecursiveCharacterTextSplitter from langchain_community.embeddings.dashscope import DashScopeEmbeddings from langchain_community.vectorstores import FAISS from langchain_community.chat_models.tongyi import ChatTongyi from langchain.prompts import PromptTemplate from langchain.chains import RetrievalQA # ============ 基本配置 ============ DOC_PATH = "knowledge_docs/knowledge.txt" INDEX_DIR = "faiss_index" EMBEDDING_MODEL = "text-embedding-v2" LLM_MODEL = "qwen-plus" CHUNK_SIZE = 500 CHUNK_OVERLAP = 100 K_VALUE = 4 def get_embedding(): """初始化Embedding模型""" return DashScopeEmbeddings(model=EMBEDDING_MODEL) def load_documents(path: str): """加载知识库文档""" loader = TextLoader(path, encoding="utf-8") return loader.load() def split_documents(documents): """文档切分""" splitter = RecursiveCharacterTextSplitter( chunk_size=CHUNK_SIZE, chunk_overlap=CHUNK_OVERLAP, separators=["\n\n", "\n", "。", ";", " ", ""] ) return splitter.split_documents(documents) def build_or_load_vector_store(documents, embedding): """构建或加载向量索引""" if os.path.exists(INDEX_DIR): print("检测到本地索引,正在加载...") return FAISS.load_local( INDEX_DIR, embedding, allow_dangerous_deserialization=True ) print("本地索引不存在,正在构建索引...") vector_store = FAISS.from_documents(documents, embedding) vector_store.save_local(INDEX_DIR) return vector_store def build_qa_chain(vector_store): """组装检索问答链""" retriever = vector_store.as_retriever( search_type="similarity", search_kwargs={"k": K_VALUE} ) template = """你是一个严谨的知识库问答助手。请仅根据下面提供的资料回答问题。 如果资料中没有对应的依据,请直接回答“根据当前知识库内容,无法找到答案”,不要编造内容。 资料: {context} 问题: {question} 请用简洁、准确的语言回答:""" prompt = PromptTemplate( input_variables=["context", "question"], template=template ) llm = ChatTongyi(model=LLM_MODEL, temperature=0.1) return RetrievalQA.from_chain_type( llm=llm, chain_type="stuff", retriever=retriever, prompt=prompt, return_source_documents=True ) def main(): # 1. 加载文档 print("正在加载文档...") docs = load_documents(DOC_PATH) print(f"共加载 {len(docs)} 个文档对象") # 2. 切分文档 chunks = split_documents(docs) print(f"切分得到 {len(chunks)} 个文本块") # 3. 初始化Embedding embedding = get_embedding() # 4. 构建或加载向量库 vector_store = build_or_load_vector_store(chunks, embedding) # 5. 构建问答链 qa_chain = build_qa_chain(vector_store) print("知识库问答已就绪,输入问题开始提问(输入 exit 或 quit 退出)") while True: question = input("\n问题:").strip() if not question: continue if question.lower() in ("exit", "quit", "退出"): break result = qa_chain.invoke({"query": question}) print(f"\n答案:{result['result']}") if "source_documents" in result: print("\n--- 参考片段 ---") for i, doc in enumerate(result["source_documents"], 1): preview = doc.page_content.strip().replace("\n", " ") print(f"[{i}] {preview[:100]}") if __name__ == "__main__": main()6.3 运行效果预览与后续扩展建议
跑通后的效果类似这样:你输入“新员工入职第一天要准备什么材料”,程序返回答案的同时,会展示答案参考了知识库里的哪几个片段。这个预览功能是我在调试阶段故意保留的,每个提问都能直接看到检索结果和生成结果的对应关系,非常有价值。
代码跑通之后,你可以继续扩展的方向包括:
- 接入更多文档格式(PDF、Word、Excel),合并多个加载器
- 把命令行交互改成Web界面,用Flask或FastAPI包一层HTTP服务
- 给不同文档打上标签,在检索时按元数据过滤,实现分类问答
- 把检索结果加上引用编号,让回答可以追溯到原文档的具体位置
我个人的体会是,RAG项目最难的从来不是链路的框架代码,而是“文档怎么切”和“检索怎么调”这两个细节。你花在调节chunk_size和看检索片段上的时间,一定会超过写代码的时间。也只有把这一步趟明白,知识库问答才会真正从“能跑”变成“能用”。