Task 4 · LangChain 应用级系统设计与 RAG 实践

Easy-Langent × DataWhale 智能体开发教程
任务笔记 · 第4章

---

本任务学什么

层级	内容
4.1	链式工作流设计（线性链 / 路由链 / 并行链）
4.2	路由链实战 + 异常处理（重试 / 捕获 / 降级）
4.3	RAG 核心原理（检索 → 增强 → 生成）
4.4	RAG 系统构建全流程（加载 → 分割 → 向量存储 → 检索生成）
学完目标	掌握"把零件组装成能用的智能机器"的能力，能独立构建完整的 RAG 问答系统

---

一、本章知识全景图

链式工作流（让大模型按步骤做事）          RAG（给大模型装知识库）
┌──────────────────────┐              ┌──────────────────────┐
│ RunnableSequence     │              │ 文档加载（TXT/PDF/Word/MD）
│ 线性链：固定顺序执行   │              │ 文本分割（Recursive...）
│                      │              │ 向量存储（FAISS）
│ RunnableBranch       │              │ 检索生成（RAG 问答）
│ 路由链：条件分发      │              └──────────────────────┘
│                      │                        ↓
│ RunnableParallel     │                  向量嵌入模型
│ 并行链：多任务同时跑   │                        ↓
└──────────────────────┘              检索策略（MMR / 相似性）
              ↓                               ↓
     异常处理（重试/捕获/降级）         RAG 评估与调优

本章核心：链式工作流解决"按步骤做事"，RAG 解决"有东西可依"，两者组合 = 能查资料、能按流程执行的智能应用。

---

二、前置知识：第三章组件的组合升级

第三章	第四章
prompt \| llm \| parser 三件套	多个三件套串联成链
Memory + Tool 组合	链式工作流 + RAG 是更复杂的组合形式
LCEL 管道符 \|	链式工作流的连接核心

---

三、链式工作流三剑客

记忆口诀：线性顺序执行、分支按条件走、并行同时干

3.1 线性链：RunnableSequence / |

适用场景：流程固定，像流水线一样一步步执行。

from langchain_openai import ChatOpenAI
from langchain_core.prompts import PromptTemplate
from langchain_core.runnables import RunnableLambda

llm = ChatOpenAI(model="deepseek-chat", temperature=0.3)

# 步骤1：提取卖点
sell_point_prompt = PromptTemplate.from_template("从以下产品介绍中提取3个核心卖点：{product_intro}")
sell_point_chain = sell_point_prompt | llm

# 格式转换（AIMessage → dict，不然下一步拿不到数据）
extract_sell_points = RunnableLambda(lambda msg: {"sell_points": msg.content})

# 步骤2：生成营销话术
marketing_prompt = PromptTemplate.from_template("根据以下卖点写营销话术：{sell_points}")
marketing_chain = marketing_prompt | llm

# 线性串联
overall_chain = sell_point_chain | extract_sell_points | marketing_chain

# 执行
result = overall_chain.invoke({"product_intro": "这款耳机续航30小时..."})

关键点：| 运算符左边输出自动作为右边输入，中间结果结构化用 RunnableLambda。

---

3.2 路由链：RunnableBranch

适用场景：一个入口、多个场景，像智能客服先分流再处理。

from langchain_core.runnables import RunnableBranch, RunnableLambda
from langchain_core.output_parsers import StrOutputParser

# 目标链1：查订单
order_chain = order_prompt | llm | StrOutputParser()

# 目标链2：退货款
refund_chain = refund_prompt | llm | StrOutputParser()

# 目标链3：保修政策
warranty_chain = warranty_prompt | llm | StrOutputParser()

# 默认链（兜底）
default_chain = default_prompt | llm | StrOutputParser()

# 路由分支
router_branch = RunnableBranch(
    (lambda x: x["scene"] == "order", order_chain),
    (lambda x: x["scene"] == "refund", refund_chain),
    (lambda x: x["scene"] == "warranty", warranty_chain),
    default_chain
)

# 完整路由链
full_chain = (
    RunnableLambda(lambda x: {"query": x["query"], "scene": router_chain.invoke({"query": x["query"]})})
    | router_branch
)

核心逻辑：RunnableBranch((条件1, 链1), (条件2, 链2), ..., 默认链) —— 匹配哪个条件就执行哪条链，都不匹配走默认链。

---

3.3 并行链：RunnableParallel / RunnableMap

适用场景：多任务同时执行，互不依赖，最后汇总结果。

from langchain_core.runnables import RunnableMap, RunnablePassthrough

# 多输入并行链
overall_chain = (
    RunnableMap({
        "sell_points": sell_point_prompt | llm | (lambda x: x.content),
        "target_audience": RunnablePassthrough(),  # 原样透传输入
    })
    | marketing_prompt | llm
)

组件	作用	类比
RunnableLambda	格式转换器，把 AIMessage 转成 dict	适配器
RunnableMap	并行执行多个 Runnable，返回 dict	多线程同时跑
RunnablePassthrough	原样传递输入	透传，不改动

---

四、异常处理三板斧

记忆口诀：临时错误靠重试，逻辑错误靠捕获，核心挂了靠降级

4.1 重试机制（解决临时性错误）

from langchain_core.runnables import RunnableRetry

retry_chain = base_chain.with_retry(
    stop_after_attempt=3,              # 最多重试3次
    wait_exponential_jitter=True,      # 指数退避 + 抖动
    retry_if_exception_type=(ConnectionError, TimeoutError),
)

两个关键概念：

概念	含义	作用
指数退避	第1次等1s，第2次等2s，第3次等4s	避免短时间内频繁请求
抖动	在指数退避基础上加随机波动（如0.8-1.2s）	避免多客户端"惊群效应"

---

4.2 异常捕获（解决可预知错误）

try:
    result = chain.invoke({"query": user_input})
except KeyError as e:
    result = fallback_chain.invoke({"query": user_input})
    print(f"输入变量缺失，已切换备用链：{e}")
except Exception as e:
    print(f"未知错误：{e}")

⚠️ 最佳实践：先捕获具体错误（KeyError），再捕获通用错误（Exception），方便调试。

---

4.3 分支降级（解决核心链失败）

from langchain_core.runnables import RunnableWithFallbacks

robust_chain = main_chain.with_fallbacks(
    fallbacks=[small_model_chain, template_chain]
)

场景：主链用 GPT-4 失败了，自动切换到用 GPT-3.5 或预设模板，保证系统可用。

---

五、RAG 核心原理

5.1 大模型的两个原生痛点

痛点	表现	根源
知识滞后	训练数据有截止日期，无法获取最新信息	大模型知识来自训练数据
事实性错误（幻觉）	一本正经胡说八道，专业场景可能致命	生成模型会"创造"看似合理的答案

RAG 解决思路：不是替代大模型，而是"赋能"——让大模型能利用外部实时、准确的知识来生成答案。

---

5.2 RAG 三步流程

检索（Retrieval）→ 增强（Augmentation）→ 生成（Generation）

用户提问 ──→ 向量数据库搜索 ──→ 相关片段 ──→ 拼接提示词 ──→ 大模型生成 ──→ 最终答案
              ↑                                                     ↓
           问题转向量                                             参考片段加持

步骤	作用	核心问题
检索	从知识库找到和问题相关的片段	找得准不准
增强	把检索结果拼进提示词	怎么拼、拼多少
生成	大模型基于检索结果生成答案	怎么利用检索结果

---

5.3 RAG 核心价值

价值	说明
提升准确性	答案有据可查，不是凭空生成
拓展知识边界	可接入实时数据、专业知识库
降低微调成本	不用为每个领域单独训练模型

---

六、RAG 系统构建四步流程

记忆口诀：加载→分割→入库→检索生成

6.1 文档加载（Document Loading）

from langchain_community.document_loaders import TextLoader, PyPDFLoader, Docx2txtLoader

# TXT（最简单）
loader = TextLoader("产品手册.txt", encoding="utf-8")
docs = loader.load()

# PDF（企业最常用）
loader = PyPDFLoader("合同.pdf")
docs = loader.load()

# Word（.docx格式）
loader = Docx2txtLoader("报告.docx")
docs = loader.load()

⚠️ 坑：PyPDFLoader 需安装 pip install pypdf，Docx2txtLoader 需安装 pip install docx2txt。

---

6.2 文本分割（Text Splitting）

两个核心原则：

• 相关性原则：分割后每个片段语义相对完整
• 大小适中原则：太长无关信息多，太短丢失上下文

from langchain.text_splitter import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=300,      # 每段300字左右
    chunk_overlap=50,    # 相邻段重叠50字，保证连续性
    separators=["。", "！", "？", "\n\n", "\n", " "]  # 按语义颗粒度从大到小
)

split_docs = splitter.split_documents(docs)

分割器	适用场景
RecursiveCharacterTextSplitter（推荐）	绝大多数场景，智能语义分割
CharacterTextSplitter	纯文本日志，简单粗暴按字数切
MarkdownTextSplitter	MD 文档，保留标题层级

---

6.3 向量存储与嵌入（Vector Storage & Embedding）

核心概念：把文字变成"数字"（向量），意思越接近的文字，向量越像。

from langchain_community.embeddings import HuggingFaceEmbeddings
from langchain_community.vectorstores import FAISS

# 嵌入模型（中文推荐）
embeddings = HuggingFaceEmbeddings(
    model_name="./models/Qwen/Qwen3-Embedding-0___6B"
)

# 向量数据库
vector_db = FAISS.from_documents(
    documents=split_docs,
    embedding=embeddings
)

# 保存本地
vector_db.save_local("faiss_db")

FAISS 优势：Meta 开源的轻量级向量数据库，无需单独部署，直接用。

---

6.4 检索与生成（Retrieval-Generation）

6.4.1 检索策略对比

策略	特点	lambda_mult	适用场景
相似性检索	返回最相似的 k 个片段	—	简单问题，精准定位
MMR 检索	兼顾相关性和多样性	0.7=偏相关，0.3=偏多样	复杂问题，需要多角度

# 相似性检索
retriever = vector_db.as_retriever(search_kwargs={"k": 3})

# MMR 检索（推荐复杂场景用）
retriever = vector_db.as_retriever(
    search_type="mmr",
    search_kwargs={"k": 3, "fetch_k": 10, "lambda_mult": 0.7}
)

lambda_mult 记忆：越接近 1 越像相似性检索（偏准），越接近 0 越追求多样性。

---

6.4.2 完整 RAG 问答链

from langchain_core.runnables import RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser

# 格式化检索结果
def format_docs(docs):
    return "\n\n".join([f"【文档{i+1}】{doc.page_content}" for i, doc in enumerate(docs)])

# RAG 问答链
rag_chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | PromptTemplate.from_template("根据以下参考信息回答问题。\n\n参考信息：{context}\n\n问题：{question}")
    | llm
    | StrOutputParser()
)

# 执行
result = rag_chain.invoke("这个产品的保修期是多久？")

---

七、RAG 系统常见问题与优化方向

7.1 常见问题

问题	原因	解决方法
检索结果不相关	分割参数不合理、嵌入模型不匹配	调整 chunk_size，换中文嵌入模型
生成答案遗漏关键信息	k 值太小，检索片段不够	增大 k 或用 MMR 检索
生成答案包含编造信息	检索片段相关性低	提升检索精度，加提示词约束
系统响应速度慢	向量数据库或模型性能差	优化向量数据库，用缓存

---

7.2 进阶优化方向

方向	核心方法
检索优化	调整分割策略、换更好的嵌入模型、优化 MMR 参数
提示词工程优化	明确要求"仅基于参考信息回答"、"无法回答则说不知道"
知识库优化	定期更新知识库、清理低质量内容、结构化知识

---

八、本章小结表

模块	解决的问题	核心组件
线性链	流程固定，多步骤顺序执行	RunnableSequence / \|
路由链	一个入口，多个场景动态分发	RunnableBranch
并行链	多任务同时执行，提高效率	RunnableParallel / RunnableMap
异常处理	临时错误 / 可预知错误 / 核心挂了	with_retry / try-except / with_fallbacks
RAG 原理	大模型知识滞后、事实性错误	检索 → 增强 → 生成
文档加载	多种格式转统一 Document	TextLoader / PyPDFLoader
文本分割	大文档变小片段，语义完整	RecursiveCharacterTextSplitter
向量存储	文本 → 向量 → 高效检索	HuggingFaceEmbeddings + FAISS
检索生成	精准找到 + 基于答案生成	相似性 / MMR + LCEL 链

---

九、自检清单

学完 Task 4，用这个清单判断是否真正掌握：

链式工作流

• 能默写三种链式工作流的名称、核心逻辑、适用场景
• 能用 | 管道符写出线性链（提取卖点→营销话术）
• 能解释 RunnableLambda 的作用（格式转换器）
• 能写出 RunnableMap + RunnablePassthrough 多输入链
• 能写出 RunnableBranch 条件路由链

异常处理

• 能解释指数退避和抖动的作用
• 能写出带重试、降级、异常捕获的健壮链

RAG 原理

• 能解释大模型的两个原生痛点（知识滞后、幻觉）
• 能说出 RAG 的三步流程（检索→增强→生成）
• 能说出 RAG 的三个核心价值

RAG 实操

• 能写出四种格式文档的加载代码
• 能解释文本分割的两个原则
• 能配置 RecursiveCharacterTextSplitter 参数
• 能写出 FAISS 向量存储代码
• 能对比相似性检索和 MMR 检索的差异
• 能用 LCEL 构建完整的 RAG 问答链

---

十、你可能没想到的几个点

1. 为什么链式工作流推荐用 | 而不是旧版 SequentialChain？

因为 | 是基于 Runnable 范式，组件之间数据传递更灵活，可以混用不同类型的组件（PromptTemplate、LLM、Parser），而旧版 SequentialChain 限制多、扩展性差。

2. 路由链的 RunnableLambda(lambda x: x) 为什么有时候是冗余的？

如果前面组件已经输出 dict 且包含 scene 字段，那 RunnableLambda(lambda x: x) 就是原样传递，显得多余。但如果需要重组数据结构或者做类型转换，它就有用了。

3. lambda_mult=0.7 为什么是常见平衡点？

因为它偏向相关性（70%），同时保留一定多样性（30%）。如果太偏向多样性（0.3），会引入无关信息；如果纯相关性（1.0），又和相似性检索没区别了。

4. RAG 的本质不是替代大模型

RAG 是"赋能"大模型，让它能利用外部知识，而不是重新训练模型。所以 RAG 适合的场景是：需要外部知识但不需要深度推理（如问答、客服）；不适合的场景是：需要创意写作、复杂推理（这些靠模型本身能力）。

5. 第四章是"承上启下"的关键章

• 承上：第三章的 Memory + Tool 组合升级为本章的"链式工作流 + RAG"
• 启下：第五章 Agent（自主决策）和第六章 LangGraph（状态机编排）都依赖本章的链式工作流基础

---

十一、连接三个底层核心

前言核心	本章关联
状态机设计	链式工作流（线性/路由/并行）= 流程编排的基础，LangGraph 的前置
工具调用逻辑	RAG 的检索-生成链 = 更复杂的"工具调用"（调用知识库）
多智能体协作	路由链的条件分发 = 多智能体协作的简化版（按条件分配任务）

---

笔记整理：Easy-Langent × DataWhale · task4任务