LangChain+通义千问搭建RAG知识库问答系统

上周有个朋友问我，公司里攒了几百份产品文档、报销制度、设备操作手册，散落在各个文件夹里，新同事想问点东西只能靠人肉搜索。他想做一个“文档问答机器人”，但一搜教程全是讲微调大模型、租GPU训练的那种，看起来就劝退。我直接给他推了另一条路：LangChain 加通义千问，做一套 RAG 知识库问答系统，半小时就能把链路跑通。这篇文章就是把当时那套可运行的代码完整拆开讲清楚，末尾附了完整版，复制下来换个 API Key 就能用。

这篇文章适合两类人：一是想给自己的资料库做智能问答、但不想碰模型训练的开发者；二是刚接触 RAG、想找一个最小可运行 Demo 来理解全流程的学习者。我不会只丢代码给你，我会把文档加载、文本切分、向量化、检索、Prompt 设计这些环节的“为什么”都讲明白。看完你会发现，RAG 这套东西原理上就是一层窗户纸，真正的坑全在细节里。

1. RAG到底在解决什么问题

1.1 大模型天生回答不了私有知识

先想一个场景：你直接问一个通用大模型“我们公司的请假审批流程是什么”，它能答吗？大概率是答不了的。因为模型在训练时只见过公开的互联网语料，它根本没见过你们公司的内部制度。这时候它又不会主动说“我不知道”，而是会根据自己理解过的类似内容，编造一套“看起来很像回事”的流程给你。这就是行业里常说的“幻觉”。

这个问题不是换一个更大的模型就能解决的。你再大的模型，它的知识截止时间也是固定的，你们的文档今天更新了，它不可能今天就学会。私有知识、企业内部资料、最新的产品变更说明，这些东西天然不在模型的训练范围里。所以大模型落地到具体业务场景时，第一个要解决的问题不是“模型不够聪明”，而是“模型看不到该看的数据”。

1.2 RAG的三段式流水线

RAG 的全称是 Retrieval-Augmented Generation，检索增强生成。这个名字很学术，但背后逻辑特别朴素——与其强迫模型记住所有知识，不如在它回答问题之前，先把相关的资料找出来递给它，让它“看着资料回答”。

整个过程分三段：

第一段是索引构建。把散落在 txt、docx、pdf 里的文档全部读取进来，切成一个个几百字的文本块，再用 Embedding 模型把每个文本块变成一组向量，最后存进向量数据库。这一步是离线完成的，跑一次就行。

第二段是检索。用户输入一个问题时，系统先把问题用同一个 Embedding 模型变成向量，然后在向量库里做相似度搜索，找出和这个问题最相关的 3 到 5 个文本块。这一步是实时的，要求快。

第三段是生成。把检索到的文本块和用户问题拼成一个 Prompt，交给大模型，让它基于这些检索到的资料作答。这一步很像开卷考试：模型不用背全部内容，但它手边有一份参考资料，回答必须从资料里找依据。

如果你想一句话记牢：RAG 就是用“开卷考试”的方式，替代“闭卷硬背”的传统问答。普通 LLM 问答是闭卷，RAG 是让模型带着参考资料进考场。

1.3 为什么我选择RAG而不是微调

很多第一次接触知识库问答的人会问：为什么不微调模型？微调确实可以把特定知识“训练进”模型参数里，但对大多数知识库场景来说，RAG 的性价比远高于微调。我拿常用的几个维度做个对比：

你可以把微调和 RAG 的关系理解为“背教材”和“查教材”。如果你要的是学生开口就是专业术语、语气完全贴合你的团队风格，那微调合适。但如果你本质上是想解决“内部资料太多了，大家找不到答案”的问题，那就不要去训练模型，而是把检索做到位。甚至很多成熟项目是先用 RAG 把知识检索链路跑通，后续再考虑用微调优化生成风格，两者并不是互斥的关系。

2. 技术选型：为什么是LangChain、通义千问、FAISS

2.1 LangChain到底帮你省了什么

没有 LangChain 的时候，做一个 RAG 系统要自己处理的事情非常多：写代码遍历目录加载文档，处理不同文件格式的解析逻辑，自己实现文本切分算法，对接向量数据库的 API，还要写 Prompt 拼接和结果解析。每一步都不难，但堆在一起就很烦。

LangChain 的核心价值是流程编排和组件标准化。它把 RAG 拆成一个个标准组件：文档加载器负责读取文件、文本分割器负责切分、Embedding 封装负责调向量模型、向量存储封装负责读写向量库、链（Chain）负责把检索结果拼进 Prompt 再调大模型。你要做的事情从“写全部代码”变成了“选组件、配参数”。

当然，LangChain 也是被吐槽最多的框架之一，因为它更新频繁、抽象层级多、版本之间破坏性改动不少。但即便如此，我依然推荐在起步阶段用它。原因很简单：RAG 的链路很固定，你没必要自己造每一块轮子。LangChain 帮你把边界和接口定义好了，等以后你真要上生产，再针对瓶颈替换具体实现也不迟。

2.2 通义千问作为大模型接口

大模型选型我选的是通义千问，代码里用的是 ChatTongyi 这个封装。选它的理由主要有三个。

第一，中文效果好。知识库问答场景里，检索到的资料和用户的问题大部分是中文，模型对中文语义的理解能力直接决定输出质量。通义千问的中文能力在实测中表现稳定，尤其是指令跟随和长文本归纳这两块。

第二，接入成本低。LangChain 社区里已经有现成的 ChatTongyi 类，填一个 API Key 和模型名就能用，不需要自己写 SDK 调用逻辑。如果你以后想换别的国产模型，也可以看 LangChain 是否支持对应的 Chat 类，接口风格基本一致。

第三，费用可控。它有不同档位的模型，日常测试用轻量型号，跑通了再升级到更强型号，切换成本几乎为零。对我这种经常要写 Demo 验证想法的人来说，这个体验很重要。

2.3 Embedding模型与向量库的选择

RAG 里有两个“向量”相关的东西容易被混为一谈：一个是把文本变成向量的 Embedding 模型，一个是存向量的向量数据库。两者都影响效果，但 Embedding 模型的影响更大——检索找不到相关内容，后面生成再好也是巧妇难为无米之炊。

Embedding 我这里用的是 text-embedding-v3，通义千问配套的中文向量模型。中文场景下，用专门优化过中文语料的 Embedding 模型，比直接用英文模型的效果好很多。这个差别在短文本检索时特别明显。

向量库我选的是 FAISS，而不是直接上 Milvus 或 Elasticsearch。原因很实在：FAISS 是一个本地计算库，pip install faiss-cpu就能用，不需要单独启动服务，不需要运维，特别适合个人项目和小规模知识库。你的知识库如果只是几千个文本块，FAISS 的检索性能绰绰有余。真到了百万级数据量，再迁移到专用向量数据库，LangChain 的接口层基本不用动，这点也是框架带来的好处。

3. 环境准备：先把地基打好

3.1 用虚拟环境隔离项目依赖

开始写代码之前，我强烈建议先建一个独立的 Python 虚拟环境。LangChain 的依赖树比一般库要复杂，它底层依赖 pydantic，而 pydantic 对版本要求非常严格。如果你把 LangChain 直接装进全局环境，大概率会跟其它项目里固定版本的 pydantic 起冲突，到时候满屏报错查得头疼。

创建命令很简单，把这段拿过去用就行：

python -m venv rag_env # Windows 激活 rag_env\Scripts\activate # macOS / Linux 激活 source rag_env/bin/activate

Python 版本建议用 3.10 或 3.11。3.8 太旧，个别新版依赖已经不再支持；3.12 也能跑，但有些包在 Windows 上可能没有预编译的 wheel，装起来麻烦。我自己的项目环境是 Python 3.11，实测最稳。

3.2 核心依赖安装清单

这是一个最小可运行方案，不需要装一大堆用不上的包。我在项目里实际安装的依赖就这几个：

pip install langchain langchain-community langchain-text-splitters pip install faiss-cpu pip install dashscope python-dotenv

逐个说下它们是干嘛的。

• langchain：主框架，提供链、Prompt 模板这些核心能力。
• langchain-community：第三方集成组件的家。我们后面要用的文档加载器、FAISS 向量库、ChatTongyi、DashScopeEmbeddings 都在这个包里。新版 LangChain 把社区集成从主包里拆出来了，所以必须单独装。
• langchain-text-splitters：文本切分工具。这个也被拆成了独立包，专门放 RecursiveCharacterTextSplitter。
• faiss-cpu：FAISS 的 CPU 版本。个人和小型知识库场景下，CPU 版本和 GPU 版本在检索速度上的差距你几乎感知不到，装 CPU 版最省事。
• dashscope：通义千问模型的官方 SDK。LangChain 的 ChatTongyi 内部会调用它，所以这也是隐性依赖。
• python-dotenv：用来读取 .env 文件里的环境变量，方便管理 API Key。

如果你的知识文档里有 PDF，再额外装一个：pip install pypdf。这是 PyPDFLoader 依赖的后端，不装的话加载 PDF 会直接报错。

3.3 API Key 配置的正确姿势

通义千问的模型服务是走 API 调用的，你需要先去模型服务控制台开通服务，然后创建一个 API Key。过程不复杂，重点是拿到 Key 之后怎么管理。

不要在代码里硬编码 API Key。很多人图快，直接把 Key 写在 main.py 里，然后整个文件夹传到 Git 仓库，Key 就泄露了。正确做法是在项目根目录建一个 .env 文件：

DASHSCOPE_API_KEY=你的key

代码里通过load_dotenv()自动读取，然后os.getenv("DASHSCOPE_API_KEY")拿到值。.env 文件记得加进 .gitignore，永远不要提交到远程仓库。

代码里还有一个健壮性细节：如果 API Key 为空，直接抛异常快速失败，而不是让程序跑到一半才因为认证失败报错。这个小习惯能省很多调试时间。

4. 核心代码拆解：一条链看懂 RAG 全流程

4.1 文档加载：把磁盘上的文件变成 Document

RAG 的第一步是把文档读进来。我用的加载器是 DirectoryLoader 加 TextLoader 的组合：

loader = DirectoryLoader( "./knowledge", glob="**/*.txt", loader_cls=TextLoader, loader_kwargs={"encoding": "utf-8"}, ) documents = loader.load()

DirectoryLoader 负责遍历一个目录，glob="**/*.txt"表示匹配目录下所有子文件夹里的 txt 文件。loader_cls=TextLoader指定具体用 TextLoader 去解析文件内容。

load() 返回的是一个 Document 对象列表。每个 Document 有两个核心字段：page_content是正文文本，metadata是文件信息，比如文件路径。这个 metadata 后面会当答案来源展示用，非常关键。

单独说明一下编码问题。TextLoader 默认按 utf-8 读取文件，但很多人电脑上的 txt 是 GBK 编码，结果一运行就抛 UnicodeDecodeError。所以我这里用 loader_kwargs 显式传了encoding="utf-8"。如果你手上的文档是 GBK，就把值改成"gbk"。最省心的做法是统一把知识库文档转成 utf-8，之后就没这个烦恼了。

4.2 文本切分：控制每个片段的信息量

文档加载进来后不能直接向量化，因为整篇文档太长了。一篇 3000 字的说明书，如果整体变成一个向量，向量里各个主题的信息会互相干扰，检索时相关性会被稀释。所以要先切块。

text_splitter = RecursiveCharacterTextSplitter( chunk_size=500, chunk_overlap=50, ) docs = text_splitter.split_documents(documents)

这里有两个参数需要重点理解：chunk_size 和 chunk_overlap。

chunk_size=500 表示每个文本块大约 500 个字符。为什么选 500？因为中文场景下一个汉字差不多对应一个 token，500 个字符就是大概 500 个 token 的文本，既能承载一个相对完整的知识点，又不会占用太多大模型上下文窗口。如果设成 2000，一个块里塞了好几个主题，检索出来的相关性就不精确；如果设成 100，每段内容太短，模型拿不到完整上下文，回答容易断章取义。

chunk_overlap=50 表示相邻两个块之间保留 50 个字符的重叠。这 50 个字符是为了防止切分边界恰好把一段完整的话拦腰截断。比如上一块结尾在讲某个概念的定义，下一块开头也是这个概念，重叠可以让两块各自都保留完整信息。

RecursiveCharacterTextSplitter 的切分逻辑不是简单地按长度硬切，而是优先按换行符、句号、空格这样的天然边界递归切分。切出来都是相对完整的句子，比无脑按字符切要智能一点。

4.3 向量化与入库：构建一次，反复使用

文本切好后，下一步就是向量化并存入向量库。这一步在我的代码里是两行：

embeddings = DashScopeEmbeddings(model="text-embedding-v3") vectorstore = FAISS.from_documents(docs, embeddings) vectorstore.save_local("./faiss_index")

第一行创建 Embedding 模型实例。第二行对每个文本块调用模型接口生成向量，并写入 FAISS 索引。第三行把索引持久化到本地磁盘。

FAISS.from_documents 是同步流程，文档多的时候这里会明显卡几十秒，是正常的，它在挨个调用 Embedding API。

有一个经验值得记下来：如果只是改动 Prompt 模板，不需要重建索引；如果你改了切分策略、换了 Embedding 模型或者新增了大量文档，就建议重新 build 一次索引。因为索引里存的是文本块的向量，Prompt 改了不影响向量内容。

4.4 检索器与Prompt设计：决定回答质量的两只手

索引构建好之后，要把它包装成检索器：

retriever = vectorstore.as_retriever( search_type="similarity", search_kwargs={"k": 4}, )

search_type="similarity"表示用向量余弦相似度找最相关的文本块。k=4表示取最相关的 4 块。这个数字不是拍脑袋定的：太小了可能漏掉关键信息，太大了又可能把一堆不太相关的内容也塞进上下文，反而干扰模型。4 这个值在大多数场景下是个不错的起点，你可以在测试时调到 3 或 5 对比效果。

Prompt 模板是整个系统里最直接的“指挥棒”，我用的模板是：

prompt_template = """请基于以下给定的资料，用简洁、专业、准确的中文回答用户的问题。 如果给定的资料中没有相关内容，请直接回答：根据现有知识无法回答该问题。禁止编造或臆测。 给定资料： {context} 用户问题： {question} 你的回答："""

这个模板的重点在于“禁止编造”四个字。RAG 系统的幻觉问题不只在检索环节，如果 Prompt 没有明确约束，模型会把检索到的资料和自己记忆里的旧知识混在一起作答。给它设置一个“不知道就直说”的兜底，是抑制幻觉最简单有效的一招。

模板里的{context}和{question}是占位符，组装时会分别替换成检索到的文本块和用户问题。

4.5 组装问答链并调用

组装用的是 RetrievalQA 这个便捷链：

qa_chain = RetrievalQA.from_chain_type( llm=llm, chain_type="stuff", retriever=retriever, return_source_documents=True, chain_type_kwargs={"prompt": QA_CHAIN_PROMPT}, ) result = qa_chain.invoke({"query": "报销流程是什么？"})

chain_type="stuff"表示把检索到的所有文本块直接拼接后塞进 Prompt。这是最简单也最推荐的方式，因为它把全文都给了模型，信息损失最小。只有当你检索结果特别多、超过模型上下文限制时，才需要考虑 map_reduce 或 refine 这样的复杂策略。

调用时输入的是一个字典，key 是query。这也是 RetrievalQA 隐式的约定，很多人第一次用会惯性写成question，结果跑起来直接报错。

输出 result 是一个字典：result["result"]是模型生成的最终回答，result["source_documents"]是命中的文本块列表，遍历它就能拿到每个块的原文和文件路径。调试时这个字段特别重要，它能帮你看到模型到底依据了哪些资料。

5. 实测中的高频坑与排查思路

5.1 中文编码引发 UnicodeDecodeError

这是最常见的启动报错。TextLoader 默认用 utf-8 读文件，但你的 txt 可能是 GBK 编码，一加载就抛 UnicodeDecodeError。

我的处理方式是：在 DirectoryLoader 里显式传loader_kwargs={"encoding": "utf-8"}，然后把知识库原始文档统一用记事本或脚本转成 UTF-8。手动改文件编码太累了，可以在代码里加个预处理，或者干脆要求团队上传文档时统一格式。如果你坚持读 GBK，就把 encoding 改成"gbk"。

5.2 FAISS加载报错：allow_dangerous_deserialization

新版本 LangChain 在加载本地 FAISS 索引时，强制要求你显式声明允许反序列化，否则会报一堆安全提示。这是 because FAISS 的本地索引是用 pickle 序列化的，反序列化恶意文件可能有安全风险。

解决办法很简单，加载时加上参数：

vectorstore = FAISS.load_local( "./faiss_index", embeddings, allow_dangerous_deserialization=True, )

这个参数看着吓人，但只要你加载的是自己生成的索引文件，在本地可信环境里跑没有实际问题。

5.3 模型调用失败：限流、超时与Key配置

如果你在循环测试时连续提问，可能会碰到服务端返回限流错误或超时。第一反应先检查几个点：API Key 是否配置在环境变量里，DASHSCOPE_API_KEY 是否真的读到了；模型名是否拼写正确；网络环境是否稳定。

排查完了如果还报限流，最简单的方案是把代码里的模型从效果更强的型号换成轻量型号。日常功能调试阶段完全够用，等确认链路没问题了再换大模型。另外可以在循环里加一个 sleep，避免一股脑发请求。

5.4 回答不对，检索内容驴头不对马嘴

遇到最多的是这个问题：代码跑通了，回答也生成了，但内容完全不在点上。

第一步先看 source_documents。把命中的几个文本块打印出来，看看模型依据的资料是不是真的和问题相关。如果检索出来的就不相关，那是索引或检索参数的问题，调整 chunk_size、换 Embedding 模型、增加 k 值。如果检索出来是相关的，但回答还是不对，那就是 Prompt 的问题，检查模板里有没有明确告诉模型“只能依据给定资料作答”。

还有一个容易被忽略的细节：测试时用文档里的原话问，先确认链路是通的；再改成同义表达，验证检索器的泛化能力。很多人一上来就用自己的大白话提问，检索不到内容就以为是系统坏了，其实在文档里原话是有明确答案的。

5.5 LangChain版本变动导致的import报错

版本兼容问题，我之前被恶心过。LangChain 从 0.1 到 0.3 经历了好几次大重构，很多组件改了导入路径。网上大量教程还是旧版写法，直接复制就会 ModuleNotFoundError。

新版的核心路径写在下面：

• RecursiveCharacterTextSplitter 从langchain_text_splitters导入
• DirectoryLoader、TextLoader、ChatTongyi、DashScopeEmbeddings、FAISS 从langchain_community子模块导入
• RetrievalQA、PromptTemplate 从langchain导入

解决版本问题有一个笨办法但很有效：遇到 ModuleNotFoundError，优先检查 import 路径是否符合上面这些新路径，而不是急着去 pip 重装。同时建议在项目 requirements.txt 里把 langchain 版本范围写死，比如langchain>=0.3,<0.4，避免哪天长版本大版本升级后旧代码突然跑不了。

6. 可直接运行的完整代码（文章末尾）

6.1 项目目录结构

先把目录结构摆出来，照着建就行：

rag_demo/ ├── .env # 存放 DASHSCOPE_API_KEY ├── knowledge/ # 把你想让模型学习的文档放这里 │ └── product_manual.txt └── main.py # 本文完整代码

首次运行 main.py 会自动在项目根目录创建 faiss_index 文件夹，用来存放向量索引。第二次再运行，如果检测到索引已存在，就不会重新构建，直接加载本地索引。

6.2 main.py 完整代码

下面这段代码就是完整可运行的版本。你只需要做三件事：把 .env 文件里的 Key 换成自己的；往 knowledge 目录里放几个 txt 文档；然后执行python main.py。

import os from dotenv import load_dotenv load_dotenv() from langchain_community.document_loaders import DirectoryLoader, TextLoader from langchain_community.embeddings import DashScopeEmbeddings from langchain_community.vectorstores import FAISS from langchain_community.chat_models import ChatTongyi from langchain_text_splitters import RecursiveCharacterTextSplitter from langchain.chains import RetrievalQA from langchain.prompts import PromptTemplate DASHSCOPE_API_KEY = os.getenv("DASHSCOPE_API_KEY") if not DASHSCOPE_API_KEY: raise ValueError("请先在 .env 文件中配置 DASHSCOPE_API_KEY") KNOWLEDGE_DIR = "./knowledge" VECTOR_INDEX_DIR = "./faiss_index" def build_index(): """读取知识库文档，切分，向量化，保存FAISS索引。""" loader = DirectoryLoader( KNOWLEDGE_DIR, glob="**/*.txt", loader_cls=TextLoader, loader_kwargs={"encoding": "utf-8"}, ) documents = loader.load() if not documents: raise ValueError("知识库目录下没有找到任何 txt 文档，请先往 knowledge 目录放文件。") text_splitter = RecursiveCharacterTextSplitter( chunk_size=500, chunk_overlap=50, ) docs = text_splitter.split_documents(documents) embeddings = DashScopeEmbeddings(model="text-embedding-v3") vectorstore = FAISS.from_documents(docs, embeddings) vectorstore.save_local(VECTOR_INDEX_DIR) print(f"索引构建完成，共切分 {len(docs)} 个文本块，已保存到 {VECTOR_INDEX_DIR}") def load_index(): """加载本地FAISS索引。""" embeddings = DashScopeEmbeddings(model="text-embedding-v3") vectorstore = FAISS.load_local( VECTOR_INDEX_DIR, embeddings, allow_dangerous_deserialization=True, ) return vectorstore def create_qa_chain(): """构建检索问答链。""" if not os.path.exists(VECTOR_INDEX_DIR): build_index() vectorstore = load_index() retriever = vectorstore.as_retriever( search_type="similarity", search_kwargs={"k": 4}, ) llm = ChatTongyi( model="qwen-plus", temperature=0.1, ) prompt_template = """请基于以下给定的资料，用简洁、专业、准确的中文回答用户的问题。 如果给定的资料中没有相关内容，请直接回答：根据现有知识无法回答该问题。禁止编造或臆测。 给定资料： {context} 用户问题： {question} 你的回答：""" QA_CHAIN_PROMPT = PromptTemplate( template=prompt_template, input_variables=["context", "question"], ) qa_chain = RetrievalQA.from_chain_type( llm=llm, chain_type="stuff", retriever=retriever, return_source_documents=True, chain_type_kwargs={"prompt": QA_CHAIN_PROMPT}, ) return qa_chain def main(): qa_chain = create_qa_chain() while True: question = input("\n请输入问题（输入 exit 或 quit 退出）：").strip() if question.lower() in ("exit", "quit"): break if not question: continue result = qa_chain.invoke({"query": question}) print("\n回答：") print(result["result"]) print("\n参考来源：") for i, doc in enumerate(result["source_documents"], 1): source = doc.metadata.get("source", "未知来源") content = doc.page_content[:120].replace("\n", " ") print(f"{i}. {source} | {content}") print("已退出。") if __name__ == "__main__": main()

6.3 运行验证与后续优化

按顺序执行：

1. 在项目根目录创建 .env 文件，写入DASHSCOPE_API_KEY=你的key。
2. 在 knowledge 目录下放几个 txt 文档，比如产品说明、FAQ、操作手册。
3. 运行python main.py。
4. 在交互提示符后输入问题，比如“备份策略是什么”，回车就能看到回答和来源。
5. 输入 exit 退出程序。

第一次运行会先构建索引，等你第二次输入问题时，检索和回答就已经是实时完成的。我实际测试下来，从索引构建到拿到第一个回答，整个过程通常不超过一分钟，而且大部分时间是花在文档切分和向量化上。

这个版本跑通之后，还有几个后续值得做的方向：如果想支持多轮对话，可以在链路里加一个记忆模块，把历史对话一并传给模型；如果知识库文档经常更新，可以在现有索引上调用 add_documents 增量添加，而不是每次全量重建；如果想摆脱在线 API 的成本，也可以把 ChatTongyi 换成本地部署的开源模型，LangChain 的接口层基本不用动。

这套代码我实际跑过很多轮，第一次跑通时最大的感受是：RAG 真的没有想象中复杂，真正的复杂度全藏在细节里——切分多大、命中几个块、Prompt 怎么写、Key 怎么配。你把这篇文章的代码复制下来，换成自己的文档跑一圈，很多“教程里为什么这么写”的问题就自然懂了。以后要做生产级系统，再把权限、日志、效果评估这些补上，但第一步永远是把链路跑通。