RAG 系列（二）：用 LangChain 搭建你的第一个 RAG Pipeline

从 100 行代码到生产级 Pipeline

上一篇我们用手写 Python 搭了一个最小 RAG，100 行代码跑通了核心逻辑。但如果你想把那套代码搬到生产环境，很快就会撞上一堵墙。

要加载 PDF？ 你需要 PyPDF2 或 pdfplumber，然后发现表格、页眉页脚的解析是一场噩梦。

要切分文本？ 你那个朴素的 text.split("\n\n") 会把句子拦腰截断、破坏代码块，或者切出超长的块直接把 Token 上限撑爆。

想换个向量数据库？ 祝你下午愉快——每个数据库的 API 都不一样，距离度量不一样，元数据处理也不一样。

想换个 LLM 提供商？ OpenAI 的客户端、Anthropic 的客户端、本地 llama.cpp……每个都有自己的消息格式、Token 计算方式、错误处理逻辑。

这正是 LangChain 存在的意义。它不搞什么魔法，也不替代底层的模型或数据库。它只做一件简单但有价值的事：给所有组件提供一个统一的接口，让你专注于 RAG 系统的业务逻辑，而不是 plumbing（ plumbing 指的是那些连接管道的脏活累活）。

这篇文章，我们用 LangChain 的现代 API 重建 RAG Pipeline。读到最后，你会拥有一个完整的、可运行的项目：它能读取 PDF、智能切分、存入 ChromaDB、用多 Provider LLM 回答问题——而真正的 Pipeline 代码只有大约 30 行。

LangChain 版本说明： 本文代码基于 langchain 1.2.x（当前主流稳定版）。langchain 1.x 对 0.3.x 做了破坏性重组，部分高层 API（如 create_retrieval_chain）被移除。代码改用 LCEL 原生语法（| 管道符）组合 Chain，功能完全等价，且不依赖版本。完整源码：github.com/chendongqi/…

RAG Pipeline 的六大组件

LangChain 把 RAG 拆解成六个组件。理解每个组件是做什么的、质量风险藏在哪里——这是以后调试 RAG 系统的关键。

把六个组件想象成工厂里的一条流水线：Document Loader 是原材料进料口，Text Splitter 是精密切割站，Embedding Model 和 Vector Store 是仓库和库存系统，Retriever 是拣货员，Chain 是车间主任——协调一切并交付最终产品。

流水线上任何一个工位配置错了，最终产品都会受影响——而棘手的是，失败看起来经常是 LLM 的问题，实际上是检索的问题。

实战：完整的 LangChain RAG 项目

我们来动手搭建。这个项目会读取一个目录下的 PDF 文件，建立索引，然后让你用自然语言提问。

项目结构

1rag-project/
2├── requirements.txt
3├── data/
4│   └── sample.pdf          # 把你的 PDF 文档放这里
5└── rag_pipeline.py
6

Step 0：安装依赖

1langchain>=0.3.0
2langchain-text-splitters>=0.3.0
3langchain-openai>=0.2.0
4langchain-chroma>=0.1.0
5langchain-community>=0.3.0
6pypdf>=4.0.0
7python-dotenv>=1.0.0
8

完整源码（可直接运行）： github.com/chendongqi/…支持智谱 AI / OpenAI / Ollama 多 LLM Provider，Embedding 支持 SiliconFlow 和本地 Ollama。

安装：

1pip install -r requirements.txt
2

还需要配置 API Key（复制 .env.example 为 .env 后填入）：

1cp .env.example .env
2# 编辑 .env，填入 LLM_API_KEY 和 EMBEDDING_API_KEY
3

支持的 Provider：

• LLM：智谱 AI（默认）、OpenAI、SiliconFlow、Ollama、Azure
• Embedding：SiliconFlow（默认）、OpenAI、Ollama

Step 1：加载文档

PyPDFLoader 帮我们处理 PDF 解析。它按页提取文本，返回一个 Document 对象列表，每个对象包含页面内容和元数据（页码、来源文件等）。

1from langchain_community.document_loaders import PyPDFLoader
2from pathlib import Path
3
4def load_pdfs(data_dir: str = "./data"):
5    """从数据目录加载所有 PDF 文件"""
6    documents = []
7    pdf_paths = list(Path(data_dir).glob("*.pdf"))
8
9    for pdf_path in pdf_paths:
10        loader = PyPDFLoader(str(pdf_path))
11        pages = loader.load()
12        documents.extend(pages)
13        print(f"已加载 '{pdf_path.name}'：{len(pages)} 页")
14
15    print(f"共加载 {len(documents)} 个文档片段")
16    return documents
17

真实世界提示： PDF 是文档格式里的"狂野西部"。如果你的 PDF 是扫描件（图片），你需要 OCR（通过 pdfplumber + Tesseract 或 Azure Document Intelligence）。如果 PDF 里有很多表格，考虑用 UnstructuredPDFLoader，它对表格结构的保留比纯文本提取好得多。

Step 2：切分文档

还记得第一篇里的分块问题吗？LangChain 的 RecursiveCharacterTextSplitter 能成为行业默认不是没道理的。它会尝试按自然边界切分——先按段落、再按换行、再按句子、再按单词——尽量避免在句子中间切断。

1from langchain_text_splitters import RecursiveCharacterTextSplitter
2
3def split_documents(documents, chunk_size=200, chunk_overlap=30):
4    """
5    将文档切分为有重叠的块
6    chunk_overlap 确保相邻块之间有上下文连续性
7    """
8    splitter = RecursiveCharacterTextSplitter(
9        chunk_size=chunk_size,
10        chunk_overlap=chunk_overlap,
11        length_function=len,
12        separators=["\n\n", "\n", ". ", " ", ""]
13    )
14
15    chunks = splitter.split_documents(documents)
16    print(f"切分为 {len(chunks)} 个块（chunk_size={chunk_size}，overlap={chunk_overlap}）")
17    return chunks
18

为什么 chunk_overlap 很重要： 如果一个关键概念横跨两个块——比如"API 速率限制是每分钟 100 次请求。超过限制会返回 429 状态码"——50 个字符的重叠能确保第二个块里还保留着"每分钟 100 次请求"的上下文。没有重叠的话，Retriever 可能只召回一个块，漏掉因果关系。

Chunk size 的权衡：

大多数场景下，512 tokens + 50 token 重叠 是一个稳妥的起点。

Step 3：Embedding 并存入 ChromaDB

现在把每个块转成向量并存起来。这里选 ChromaDB，因为它支持持久化（数据不会随重启丢失）、支持元数据过滤、而且零配置——作为嵌入式数据库本地运行。

1from langchain_openai import OpenAIEmbeddings
2from langchain_chroma import Chroma
3import os
4
5persist_directory = "./chroma_db"
6
7def build_vector_store(chunks):
8    """创建 Embedding 并存入 ChromaDB"""
9    embeddings = OpenAIEmbeddings(
10        model="BAAI/bge-large-zh-v1.5",  # SiliconFlow 中文模型
11        api_key=os.getenv("EMBEDDING_API_KEY"),
12        base_url="https://api.siliconflow.cn/v1",
13        dimensions=1024,
14        chunk_size=32  # SiliconFlow 限制每批最多 32 条
15    )
16
17    if os.path.exists(persist_directory):
18        import shutil
19        shutil.rmtree(persist_directory)
20
21    vector_store = Chroma.from_documents(
22        documents=chunks,
23        embedding=embeddings,
24        persist_directory=persist_directory
25    )
26    print(f"向量库构建完成：{vector_store._collection.count()} 个向量已持久化")
27    return vector_store
28

Embedding 模型说明： 这里用 SiliconFlow 上的 BAAI/bge-large-zh-v1.5（中文效果优秀），通过 langchain_openai 的 OpenAI 兼容接口接入。如果用 OpenAI 官方，换成 text-embedding-3-small 即可。chunk_size=32 是 SiliconFlow 的批次限制（每批最多 32 条），其他 Provider 通常默认 1000 条。

Step 4：构建 Retriever

Retriever 是向量库的一个薄封装，负责处理搜索逻辑。默认情况下，它做相似度搜索，返回最相关的 Top-K 个块。

1def get_retriever(vector_store, search_k=4):
2    """配置相似度检索的 Retriever"""
3    retriever = vector_store.as_retriever(
4        search_type="similarity",
5        search_kwargs={"k": search_k}
6    )
7    return retriever
8

Step 5：构建 RAG Chain

这里是 LangChain 现代 API 最亮眼的地方。不用老旧的 RetrievalQA 类，我们用 LCEL（LangChain Expression Language，LangChain 表达式语言）来组合 Chain——代码更直观、更容易调试、也更容易修改。

1from langchain_openai import ChatOpenAI
2from langchain_core.prompts import ChatPromptTemplate
3from langchain_core.output_parsers import StrOutputParser
4from langchain_core.runnables import RunnablePassthrough
5
6def build_rag_chain(retriever):
7    """
8    构建完整 RAG Chain（langchain 1.x 兼容写法，不依赖 create_retrieval_chain）
9    检索 → format_docs → 塞进 Prompt → LLM → StrOutputParser
10    """
11    llm = ChatOpenAI(
12        model="glm-4-flash",  # 智谱 AI 模型，通过 SiliconFlow 或直连
13        api_key=os.getenv("LLM_API_KEY"),
14        base_url="https://open.bigmodel.cn/api/paas/v4",
15        temperature=0
16    )
17
18    # System Prompt，{context} 由 format_docs 填充，{question} 是用户原始问题
19    system_prompt = (
20        "你是一个精准的知识助手。请仅根据下方提供的参考内容回答用户问题。"
21        "如果参考内容中没有答案，请明确说明——不要编造。\n\n"
22        "参考内容：\n{context}"
23    )
24
25    prompt = ChatPromptTemplate.from_messages([
26        ("system", system_prompt),
27        ("human", "{question}")
28    ])
29
30    # 辅助函数：把 Document 列表转成字符串
31    def format_docs(docs: list) -> str:
32        return "\n\n".join(doc.page_content for doc in docs)
33
34    # LCEL Chain：用管道符 | 组合各个组件
35    # 1. {"context": retriever | format_docs, "question": RunnablePassthrough()}
36    #    → retriever 检索文档，format_docs 把 Document 对象列表转成字符串
37    # 2. | prompt → 组装成完整的 Prompt
38    # 3. | llm → LLM 生成答案
39    # 4. | StrOutputParser() → 输出纯文本（而不是 AIMessage 对象）
40    rag_chain = (
41        {
42            "context": retriever | format_docs,
43            "question": RunnablePassthrough()
44        }
45        | prompt
46        | llm
47        | StrOutputParser()
48    )
49
50    return rag_chain
51

这里发生了什么？ 我们用的是 LCEL（LangChain Expression Language）原生语法，用 | 管道符把各个组件串联起来，而不是用高层的 create_retrieval_chain（该函数在 langchain 1.x 中已被移除）。

关键在于 retriever | format_docs：Retriever 返回的是 Document 对象列表，format_docs 把它转成字符串填入 {context} 占位符。RunnablePassthrough() 把用户的原始问题透传到 {question} 占位符。这三行代码等价于第一篇里的手写检索 + 组装 + 生成逻辑。

Step 6：查询 Pipeline

1def query(rag_chain, question: str, retriever):
2    """把问题丢进 RAG Pipeline 跑一遍，打印答案和来源"""
3    print(f"\n问题：{question}")
4
5    answer = rag_chain.invoke(question)  # LCEL Chain 直接返回纯文本
6    print(f"\n答案：\n{answer}")
7
8    # 单独检索一次来源（rag_chain 不直接暴露 retrieved docs）
9    docs = retriever.invoke(question)
10    print("\n检索到的来源：")
11    for i, doc in enumerate(docs, 1):
12        source = doc.metadata.get("source", "未知")
13        page = doc.metadata.get("page", "?")
14        preview = doc.page_content[:120].replace("\n", " ")
15        print(f"  [{i}] {source}（第 {page} 页）：{preview}...")
16
17    return answer
18

组装起来

1if __name__ == "__main__":
2    # 1. 加载
3    docs = load_pdfs("./data")
4
5    # 2. 切分（PDF 每页内容短，用较小的 chunk_size）
6    chunks = split_documents(docs, chunk_size=200, chunk_overlap=30)
7
8    # 3. Embedding & 存储
9    vector_store = build_vector_store(chunks)
10
11    # 4. 检索器
12    retriever = get_retriever(vector_store, search_k=4)
13
14    # 5. 构建 Chain（LCEL 方式）
15    rag_chain = build_rag_chain(retriever)
16
17    # 6. 交互式提问
18    while True:
19        user_input = input("\n你的问题（输入 quit 退出）：").strip()
20        if user_input.lower() in ("quit", "exit", "q"):
21            break
22        if user_input:
23            query(rag_chain, user_input, retriever)
24

运行 Pipeline

把 PDF 放到 data/sample.pdf，然后运行：

1python rag_pipeline.py
2

示例输出：

1RAG Pipeline 启动
2  LLM Provider    : zhipu
3  LLM Model       : glm-4-flash
4  Embedding       : openai / BAAI/bge-large-zh-v1.5
5  数据目录        : ./data
6  向量库          : ./chroma_db
7==================================================
8已加载 'Automotive-SPICE-PAM-v40.pdf'：153 页
9共加载 153 个文档片段
10切分为 2000 个块（chunk_size=200，overlap=30）
11[Embedding] Provider: openai | Model: BAAI/bge-large-zh-v1.5 | Base: https://api.siliconflow.cn/v1
12已清除旧向量库：./chroma_db
13向量库构建完成：2000 个向量已持久化
14==================================================
15RAG Pipeline 构建完成！输入问题开始问答（输入 'quit' 退出）
16==================================================
17
18你的问题：什么是 Automotive SPICE？
19
20==================================================
21问题：什么是 Automotive SPICE？
22==================================================
23
24答案：
25Automotive SPICE（Automotive Software Process Improvement and
26Capability Determination）是一种用于评估和改进汽车软件
27开发过程能力的框架。它定义了软件开发生命周期中的关键
28过程域，并建立了过程能力的等级评估标准...
29
30检索到的来源：
31  [1] ./data/Automotive-SPICE-PAM-v40.pdf（第 5 页）：Automotive
32      SPICE Process Assessment Model The Process Assessment Model
33      (PAM) defines the processes...（正文来自 RAG 实际运行）
34

注意答案里带有来源引用——我们清楚知道信息来自哪几页。这种可追溯性对生产级 RAG 系统至关重要，因为用户需要验证答案的可靠性。

和 100 行手写版相比，变了什么？

对比一下第一篇的手写 RAG 和现在的 LangChain Pipeline：

这些抽象没有隐藏复杂性——而是隔离了它。当你需要调试检索质量时，你知道该调哪个组件。当你想换更便宜的 Embedding 模型时，改一行代码。当你的数据量超出 ChromaDB 的能力时，切到 Qdrant 不用动 Pipeline 的其他部分。

关于 LangChain 版本兼容性： 本文的代码基于 langchain 1.x（当前稳定版）。langchain 1.x 对 0.3.x 做了破坏性重组，create_retrieval_chain 和 create_stuff_documents_chain 在 1.x 中已被移除。代码用的是 LCEL 原生语法（|> 管道符组合），功能完全等价，且不依赖特定版本的高层 API。

每个阶段的常见坑

Loader 坑："我的 PDF 里有表格，解析出来是一团糟"

原始 PDF 文本提取会把表格拍扁成一串数字。表格多的文档，用 UnstructuredPDFLoader 或 AzureAIDocumentIntelligenceLoader，它们能保留结构关系。

Splitter 坑："答案被切成了两半，模型只看到了一半"

把 chunk_overlap 提高到 100-150，或者减小 chunk_size 让关键概念能放进一个块。更好的方案是 Parent-Document Retrieval（后续文章会讲）——用小块做检索，但返回完整的父文档作为上下文。

Embedding 坑："问题和文档明明相关，就是匹配不上"

这是"非对称检索"问题。用户问"怎么重置密码？"，文档写的是"要重置密码，请前往 设置 → 安全"。问题和答案的表面文本差异大，向量距离也远。解法：用针对 Q&A 检索微调的模型（如 BGE-M3），或者生成假设答案来做检索（HyDE——后续也会讲）。

Retriever 坑："Top-K=4 不够用，复杂问题漏信息"

如果一个问题需要综合文档里五个不同部分的信息，k=4 就会漏掉一个。但盲目增大 k 会引入噪声。更好的做法：Multi-Query Retrieval（生成 3 个问题的变体，分别检索，去重）或者 Reranking（先召回 20 个，再用 Cross-Encoder 挑出最好的 5 个）。

Chain 坑："模型无视上下文，开始 hallucinate"

Prompt 设计很重要。系统 Prompt 必须明确指示模型只用提供的上下文回答。加上"如果参考内容中没有答案，请明确说明——不要编造"能显著提升忠实度。我们会在评估篇里用 RAGAS 定量测量这个指标。

小结

这篇文章把第一篇的裸奔 RAG 概念，包上了一个生产级框架。我们讲了：

1. LangChain RAG Pipeline 的六大组件——Loader、Splitter、Embedding、Vector Store、Retriever、Chain，每个组件藏了什么质量风险。
2. 一个完整可运行的项目——加载 PDF、用 RecursiveCharacterTextSplitter 切分、OpenAI Embedding、ChromaDB 存储、LangChain LCEL Chain 问答。
3. Chunk size 的权衡——实际项目中 PDF 每页内容可能很短（200 字符），这时候 512 的默认值会产生 0 个块。200 + 30 重叠是实测可用的安全值。
4. 每个阶段的常见坑——从 PDF 表格解析失败，到非对称检索失配。

这篇文章的代码是一个扎实的基础。它能处理真实 PDF、持久化数据、还能给出来源引用。但它仍然是一个朴素 RAG（Naive RAG）——一次查询、一次检索、一次回答。后面的文章会逐步加入区分玩具 Demo 和生产系统的组件：混合检索、Reranking、查询优化、评估框架。

参考资料

• LangChain RAG 教程 —— 官方 RAG 快速入门
• LangChain Expression Language (LCEL) —— 为什么以及如何用 LCEL 构建可组合 Chain
• ChromaDB 文档 —— 向量库配置、持久化与查询

《RAG 系列（二）：用 LangChain 搭建你的第一个 RAG Pipeline》 是转载文章，点击查看原文。