ARTICLE · 1144450
LangChain 综合项目实战:PDF 私有知识库 RAG 问答,从零搭一套企业级智能问答系统
别再让你的 PDF 文档吃灰了!用 LangChain + ChromaDB,30 分钟搭建一套能真正回答业务问题的私有知识库。
你是否遇到过这样的场景:
• 公司内部的制度手册、技术文档全在 PDF 里,员工找不到想要的信息 • 客户咨询的问题重复率高达 70%,客服每天都在复制粘贴 • 几百页的产品手册,搜关键词翻半天还是定位不到答案
传统方案要么建搜索系统(贵且慢),要么靠人工(效率低还易出错)。
有没有一种方式,让 AI 直接读懂你的文档,然后基于真实内容给出精准回答?
答案是:RAG(检索增强生成)。
今天我们就用最流行的 LangChain 框架,从 0 到 1 搭建一套完整的 PDF 私有知识库问答系统。全程代码可跑,跟着做就能上线!
一、什么是 RAG?为什么它能解决你的问题?
1.1 RAG 的核心思路
RAG(Retrieval Augmented Generation)的核心理念很简单:
传统 LLM RAG┌─────────┐ ┌──────────┐│ 用户提问 │────▶│ 检索引擎 ││ │ │ 从知识库 ││"XX是什么"│ │ 找出相关内容│AI直接回答│ │ ││(可能编造)│ │ ▼ │└─────────┘ │LLM+上下文 │ │基于文档回答│ └──────────┘核心差异:
• ❌ 传统 AI:只能回答训练数据里见过的内容,知识截止日期后的事一问三不知,还会"一本正经地编" • ✅ RAG 系统:先从你自己的文档中检索相关内容,再把内容交给 AI 让它基于事实作答,有据可查
1.2 RAG vs 微调:选哪个?
| 数据更新 | ||
| 知识边界 | ||
| 实现成本 | ||
| 幻觉控制 | ||
| 可追溯性 | ||
| 适用场景 |
建议:大多数知识问答场景,RAG 是首选方案。微调更适合需要改变模型行为模式(比如让模型学会特定格式输出)的场景。
二、LangChain RAG 系统架构
一个完整的 LangChain RAG 应用可以串成一条清晰的数据流:
原始 PDF 文档 │ ▼┌──────────────────┐│ Document Loader │ ← 文档加载器:PDF/Word/网页...│ (PyPDFLoader) │└────────┬─────────┘ │ Document(整篇文本 + 元数据) ▼┌──────────────────┐│ Text Splitter │ ← 文本分割器:切块、重叠│ (Recursive...) │└────────┬─────────┘ │ chunks(小块文本,继承元数据) ▼┌──────────────────┐│ Embedding Model │ ← 嵌入模型:文字→向量│ (BGE/OpenAI) │└────────┬─────────┘ │ 向量 → 写入数据库 ▼┌──────────────────┐│ Vector Store │ ← 向量数据库:持久化存储│ (Chroma/FAISS) │└────────┬─────────┘ │ 用户提问同样转向量,相似度检索 top-k ▼┌──────────────────┐│ Retriever │ ← 检索器:找到最相关的 chunks└────────┬─────────┘ │ 问题 + 相关 chunks(拼进 Prompt) ▼┌──────────────────┐│ LLM (生成) │ ← 大模型基于文档回答│ (Qwen/GPT...) │└────────┬─────────┘ ▼ 基于文档的回答 + 来源引用六大阶段,环环相扣——这就是 RAG 的完整链路。接下来我们一步步实现它。
三、环境准备
3.1 安装依赖
# 创建虚拟环境(强烈推荐!避免包冲突)python -m venv rag-envsource rag-env/bin/activate # Windows: rag-env\Scripts\activate# 一键安装所有核心依赖pip install langchain langchain-community langchain-openai \ chromadb pypdf sentence-transformers tiktoken核心依赖说明:
langchain | ||
langchain-community | ||
langchain-openai | ||
chromadb | ||
pypdf | ||
sentence-transformers |
3.2 API Key 配置
# 创建 .env 文件cat > .env << 'EOF'# OpenAI(兼容接口,也支持通义千问等)OPENAI_API_KEY=sk-xxx# 可选:自定义 base_url(如使用国内服务)# OPENAI_BASE_URL=https://api.xxx.com/v1EOFpip install python-dotenv⚠️ 版本迁移注意:LangChain 2.0+ 拆分了模块包。旧的
from langchain.text_splitter现在要从langchain_text_splitters导入,旧写法已废弃。
四、完整代码实现(6 步走)
Step 1:文档加载 — 从 PDF 到结构化文本
from langchain_community.document_loaders import PyPDFLoader# 加载单个 PDFloader = PyPDFLoader("docs/handbook.pdf")documents = loader.load()print(f"共加载 {len(documents)} 页")print(f"每页内容: Document(page_content='{documents[0].page_content[:50]}...', metadata={documents[0].metadata})")PyPDFLoader 输出的是什么?
每个 Document 对象包含:
• .page_content— 该页的纯文本内容• .metadata— 元数据(来源文件、页码等)
💡 加载多个 PDF:用
DirectoryLoader可批量读取目录下所有 PDFfrom langchain_community.document_loaders import DirectoryLoaderloader = DirectoryLoader("docs/", glob="/**/*.pdf", loader_cls=PyPDFLoader)documents = loader.load()
Step 2:文本分割 — 把大文档切成小块
为什么要分块? LLM 有上下文窗口限制,不可能一次塞进 200 页的 PDF。更重要的是,小块内容可以独立检索——用户问"报销流程是什么",只需要返回包含报销的那几段,而不是整份手册。
from langchain_text_splitters import RecursiveCharacterTextSplittersplitter = RecursiveCharacterTextSplitter( chunk_size=1000, # 每块约 1000 字符 chunk_overlap=200, # 相邻块重叠 200 字符,防止信息被切断 add_start_index=True # 记录每块在原文档中的起始位置(方便溯源))chunks = splitter.split_documents(documents)print(f"分割为 {len(chunks)} 个 chunk")RecursiveCharacterTextSplitter 的工作原理:
按优先级依次尝试不同的分隔符切割:段落(\n\n) → 换行(\n) → 句号(./。) → 空格() → 字符
这样能尽量保持语义完整性,避免在句子中间生硬截断。
Step 3:Embedding — 文字向量化
Embedding 模型的作用:把文本转化为高维向量。语义越相似的两个文本,其向量在空间中距离越近——这就是 RAG 检索的数学基础。
# 方案 A:OpenAI Embedding(开箱即用)from langchain_openai import OpenAIEmbeddingsembeddings = OpenAIEmbeddings(model="text-embedding-3-small")# 方案 B:本地 BGE 模型(中文场景首选,无需 API)from langchain_huggingface.embeddings import HuggingFaceEmbeddingsembeddings = HuggingFaceEmbeddings( model_name="BAAI/bge-large-zh-v1.5", # 中文 RAG 默认推荐)Embedding 模型选型指南:
text-embedding-3-small | ||||
text-embedding-3-large | ||||
bge-large-zh-v1.5 | 中文 RAG 默认首选 | |||
bge-m3 | ||||
Qwen3-Embedding |
💡 推荐:中文场景首选
bge-large-zh-v1.5——质量、成本、生态三者的最佳平衡。需要处理超长文档或混合检索时升级为bge-m3。
Step 4:向量数据库 — 持久化存储
把向量化后的文本存入向量数据库,供后续快速检索。
from langchain_chroma import Chromaimport chromadb# 创建持久化的 Chroma 向量库chroma_client = chromadb.PersistentClient(path="./chroma_db")vectorstore = Chroma( collection_name="pdf_kb", embedding_function=embeddings, client=chroma_client,)# 将 chunks 写入向量库(一次性操作,后续直接读取)from langchain_core.utils import merge_dictsvectorstore.add_documents(chunks)print(f"已存储 {vectorstore._collection.count()} 条向量")向量数据库选型建议:
| Chroma | |||
| FAISS | |||
| Qdrant | |||
| Milvus |
💡 选型路径:Chroma 起步 → Qdrant 规模化 → Milvus 企业进阶。10 万条向量以下,Chroma 完全够用。超过 500 万再考虑 Milvus。
Step 5:检索 — 从知识库找答案
# 从已存储的向量库创建检索器retriever = vectorstore.as_retriever( search_type="similarity", # 相似度检索(默认) search_kwargs={"k": 5}, # 返回最相关的 5 个 chunk)# 测试检索效果query = "公司报销流程是什么?"relevant_docs = retriever.invoke(query)for i, doc in enumerate(relevant_docs): print(f"[{i+1}] 来源: {doc.metadata.get('source', '未知')} | 页码: {doc.metadata.get('page', '?')}") print(f" 内容预览: {doc.page_content[:100]}...") print()检索策略升级路径:
similarity | ||
mmr | ||
hybrid | ||
rerank |
💡 低成本升级:检索结果重复单一时,先把
search_type换成"mmr"。想进一步提升精度,加一层重排序(BGE Reranker / Cohere Rerank),效果立竿见影。
Step 6:生成 — LLM 基于文档作答
用 LangChain 的 LCEL(LangChain Expression Language) 管道语法,把检索和生成串成一条完整的链:
from langchain_openai import ChatOpenAIfrom langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholderfrom langchain_core.output_parsers import StrOutputParserfrom langchain_core.runnables import RunnablePassthrough# 1. 定义提示词模板prompt_template = ChatPromptTemplate.from_messages([ ("system", """你是一个智能助手,请严格基于以下参考资料回答用户问题。要求:1. 只使用参考资料中的信息,不要编造2. 如果参考资料中找不到答案,直接说"抱歉,资料中没有相关信息"3. 引用请注明出处(文件/页码)"""), ("human", "参考资料:\n{context}\n\n问题:{question}"),])# 2. 定义大模型llm = ChatOpenAI(model="gpt-4o", temperature=0)# 3. 组装 RAG 链(LCEL 管道语法)rag_chain = ( {"context": retriever, "question": RunnablePassthrough()} | prompt_template | llm | StrOutputParser())# 4. 运行问答!answer = rag_chain.invoke("公司报销流程是什么?")print(answer)LCEL 管道语法 | 的理解:
| 就是数据流管道,左边输出作为右边输入:
检索器 → 提示词模板 → LLM → 输出解析 │ │ │ │ ▼ ▼ ▼ ▼上下文+问题 → 格式化prompt → AI生成 → 纯文本答案五、完整项目文件结构
rag-project/├── docs/ # PDF 知识库目录│ ├── handbook.pdf│ ├── policy.pdf│ └── ...├── chroma_db/ # 向量数据库(自动创建)├── main.py # 主入口:索引构建├── qa.py # 问答接口├── requirements.txt # 依赖清单└── .env # API Key 配置main.py — 一键索引构建:
"""PDF 知识库 RAG 系统 - 索引构建脚本用法: python main.py"""from langchain_community.document_loaders import PyPDFLoader, DirectoryLoaderfrom langchain_text_splitters import RecursiveCharacterTextSplitterfrom langchain_openai import OpenAIEmbeddingsfrom langchain_chroma import Chromaimport chromadbdef build_index(pdf_dir="./docs", db_path="./chroma_db"): # 1. 加载所有 PDF loader = DirectoryLoader(pdf_dir, glob="/**/*.pdf", loader_cls=PyPDFLoader) documents = loader.load() print(f"📄 加载了 {len(documents)} 页文档") # 2. 文本分块 splitter = RecursiveCharacterTextSplitter( chunk_size=1000, chunk_overlap=200, add_start_index=True ) chunks = splitter.split_documents(documents) print(f"✂️ 分割为 {len(chunks)} 个文本块") # 3. Embedding + 存入向量库 embeddings = OpenAIEmbeddings(model="text-embedding-3-small") chroma_client = chromadb.PersistentClient(path=db_path) vectorstore = Chroma( collection_name="pdf_kb", embedding_function=embeddings, client=chroma_client, ) vectorstore.add_documents(chunks) print(f"✅ 索引构建完成,共 {vectorstore._collection.count()} 条向量")if __name__ == "__main__": build_index()qa.py — 交互式问答:
"""PDF 知识库 RAG 系统 - 问答接口用法: python qa.py"""from langchain_openai import ChatOpenAI, OpenAIEmbeddingsfrom langchain_chroma import Chromaimport chromadbfrom langchain_core.prompts import ChatPromptTemplatefrom langchain_core.output_parsers import StrOutputParserfrom langchain_core.runnables import RunnablePassthroughdef setup_rag(db_path="./chroma_db"): embeddings = OpenAIEmbeddings(model="text-embedding-3-small") chroma_client = chromadb.PersistentClient(path=db_path) vectorstore = Chroma( collection_name="pdf_kb", embedding_function=embeddings, client=chroma_client, ) retriever = vectorstore.as_retriever(search_kwargs={"k": 5}) prompt = ChatPromptTemplate.from_messages([ ("system", "基于以下资料回答:\n{context}\n\n问题:{question}"), ("human", "{question}"), ]) llm = ChatOpenAI(model="gpt-4o", temperature=0) chain = ( {"context": retriever, "question": RunnablePassthrough()} | prompt | llm | StrOutputParser() ) return chainif __name__ == "__main__": rag = setup_rag() print("🤖 RAG 问答系统已启动!输入 q 退出\n") while True: question = input("\n❓ 你的问题:") if question in ("q", "quit"): break answer = rag.invoke(question) print(f"\n💡 回答:\n{answer}")六、关键参数调优指南
6.1 chunk_size 怎么选?
| 1000 | 200 | 通用场景(推荐默认值) | 平衡精度和覆盖度 |
💡 经验法则:
chunk_overlap设为chunk_size的 10-20%。太小容易切断关键信息,太大则浪费 token。
6.2 top-k(检索数量)怎么选?
• k=3~5:大多数场景够用,返回最相关的几段 • k=8~10:搭配重排序器使用,先多召回再精排 • k>10:通常不建议,因为 LLM 上下文窗口有限
6.3 提示词模板优化
好的 System Prompt 能显著降低幻觉率:
# ✅ 推荐写法system_prompt = """你是一个专业助手。请严格基于以下参考资料回答问题:1. 只使用资料中的信息,不要编造或推测2. 如果资料中没有答案,直接说"抱歉,资料中未找到相关信息"3. 引用请注明出处(文件名和页码)4. 回答要简洁、条理清晰"""# ❌ 避免的写法system_prompt = "请回答以下问题:" # 太松泛,容易导致幻觉七、进阶:检索优化三板斧
当基础 RAG 效果不理想时,按以下顺序尝试升级:
🔪 第一板斧:MMR 检索(零成本)
# 只需改一行!把 similarity 换成 mmrretriever = vectorstore.as_retriever( search_type="mmr", search_kwargs={"k": 5, "lambda_mult": 0.7},)lambda_mult 调参建议:0.5(最大多样性)→ 0.7(推荐默认值)→ 1.0(纯相似度)
🔪 第二板斧:混合检索(向量 + 关键词)
对于产品编号、专有名词等精确匹配场景,纯向量检索效果不佳。加入 BM25 关键词检索:
from langchain.retrievers import EnsembleRetrieverfrom langchain_community.retrievers import BM25Retriever# 向量检索器vector_retriever = vectorstore.as_retriever(search_kwargs={"k": 5})# 关键词检索器bm25_retriever = BM25Retriever.from_documents(chunks)bm25_retriever.k = 5# 混合检索:融合两种结果ensemble_retriever = EnsembleRetriever( retrievers=[vector_retriever, bm25_retriever], weights=[0.7, 0.3], # 向量权重70%,关键词30%)🔪 第三板斧:重排序(Cross-Encoder 精排)
初筛返回 10 条候选,用 Cross-Encoder 模型精准打分,选出 Top 3 给 LLM:
from langchain.retrievers.context_compressor import ContextualCompressionRetrieverfrom langchain.retrievers.document_compressors import CrossEncoderRerankerfrom langchain_community.cross_encoders import HuggingFaceCrossEncoder# 加载重排序模型(本地运行,无需 API)model = HuggingFaceCrossEncoder(model_name="BAAI/bge-reranker-base")compressor = CrossEncoderReranker(model=model, top_n=3)# 将重排序包装到检索器上compression_retriever = ContextualCompressionRetriever( base_retriever=vectorstore.as_retriever(search_kwargs={"k": 10}), base_compressor=compressor,)# 后续用法不变,只是把 retriever 替换成 compression_retrieverrag_chain = ( {"context": compression_retriever | (lambda docs: "\n".join([d.page_content for d in docs])), "question": RunnablePassthrough()} | prompt_template | llm | StrOutputParser())💡 效果对比:加入重排序后,Top-3 召回率通常能提升 15-30%,答案准确度明显改善。代价是多一次推理延迟(约 100-200ms)。
八、常见问题 FAQ
Q1:PDF 解析出来是乱码怎么办?
原因:某些 PDF 使用了特殊字体或加密,pypdf 无法正确提取文字。
解决:
# 方案 A:改用 unstructured(支持更多格式)from langchain_community.document_loaders import UnstructuredPDFLoaderloader = UnstructuredPDFLoader("docs/handbook.pdf")# 方案 B:扫描件 PDF,先用 OCR 转文字pip install pytesseractQ2:检索出来的内容张冠李戴?
原因:Embedding 模型在向量空间中过于接近,导致误匹配。
解决:改用领域专用的 Embedding 模型(如中文用 bge-large-zh-v1.5),或加入混合检索 + 重排序。
Q3:回答总是"资料中没有相关信息"?
排查顺序:
1. 检查文档是否真的包含答案 → 手动搜索确认 2. 增大 k值(从 5 调到 10)→ 扩大召回范围3. 调大 chunk_size→ 避免关键信息被切散4. 更换 Embedding 模型 → 尝试 BGE 系列
Q4:向量数据库要持久化吗?
必须持久化! 否则每次启动都要重新 Embedding,浪费时间又浪费 API 费用。
# ✅ 正确的做法:PersistentClient + 指定路径chroma_client = chromadb.PersistentClient(path="./chroma_db")# ❌ 错误的做法:内存模式,重启就没了vectorstore = Chroma(embedding_function=embeddings)Q5:生产环境用什么向量数据库?
九、总结
今天我们从零搭建了一套完整的 PDF 私有知识库 RAG 问答系统:
1. 文档加载 — PyPDFLoader把 PDF 变成结构化文本2. 文本分块 — RecursiveCharacterTextSplitter保持语义完整性3. Embedding — 文字向量化,中文推荐 BGE,英文用 OpenAI 4. 向量存储 — Chroma 本地持久化,生产环境切换 Qdrant/Milvus 5. 检索 — 相似度搜索起步,按需升级 MMR → 混合检索 → 重排序 6. 生成 — LCEL 管道语法,简洁优雅地串联全链路
核心要点:
• ✅ RAG 是知识库问答的首选方案——实时更新、有据可查、不会编造 • ✅ chunk_size=1000, overlap=200是通用场景的默认黄金参数• ✅ 中文 Embedding 首选 bge-large-zh-v1.5,超长文档用bge-m3• ✅ 向量数据库按规模选:Chroma → Qdrant → Milvus • ✅ 检索优化三板斧:MMR(零成本)→ 混合检索 → 重排序
下一步可以做的事:
• 加 UI:用 Streamlit/Gradio 做出问答界面,15 行代码搞定 • 加权限:按部门/角色过滤可访问的文档 • 加评估:用 Ragas 工具量化答案质量,持续迭代 • 扩展格式:Word、Excel、网页——换加载器即可
完整的 RAG 项目代码,欢迎动手试试! 🚀
📢 关注"大强哥爱编程"公众号,获取更多AI技术前沿资讯!
🌟 喜欢这篇文章?请点赞、转发、收藏!有任何问题或建议,欢迎在评论区留言交流!