彻底搞懂 LangChain 链式调用：从 RunnableParallel 到 RAG 实战

摘要：还在被 LangChain 的 | 符号和 RunnableParallel 绕晕？本文通过图解数据流，带你彻底理解 RAG 链式调用的核心逻辑，从此不再“魔法编程”。

最近在复盘一个 RAG（检索增强生成）项目时，很多开发者对 LangChain 的链式调用（Chain）表示“头大”。特别是下面这段代码，看起来像天书：

retrieval_chain = (
    RunnableParallel(
        context=vec_store.as_retriever() | format_docs,
        question=RunnablePassthrough()
    )
    | prompt 
    | llm 
    | StrOutputParser()
)

灵魂三问：

为什么要把 question 再传一遍？
RunnableParallel 到底输出了什么？
数据是怎么在这些组件之间流动的？

今天，我们就把这段代码拆碎了，揉烂了，彻底搞懂它！

01 核心隐喻：工厂流水线

把 LangChain 的链式调用想象成一条工厂流水线。

原料：用户的提问（String）。
工序：各个组件（Retriever, Prompt, LLM…）。
操作符 |：传送带，把上一个工序的成品传给下一个工序。

1	[用户提问] ==> (检索) ==> (提示词) ==> (LLM) ==> [最终答案]

但在 RAG 场景下，有个特殊需求：提示词（Prompt）需要两份原料。

用户的问题（原始输入）
检索到的上下文（经过向量库查询得到）

可是，流水线的入口只有一份原料（用户问题）。怎么变出两份？

这就是 RunnableParallel 登场的原因。

02 拆解 RunnableParallel：分流与合并

RunnableParallel 的作用可以理解为**“并行处理，结果打包”**。

它会把输入的数据，同时交给多个分支处理，最后把结果打包成一个字典（Dictionary）。

代码映射

RunnableParallel(
    context=vec_store.as_retriever() | format_docs,  # 分支 A
    question=RunnablePassthrough()                   # 分支 B
)

数据流图解

假设用户输入是 "什么是 RAG？"

输入："什么是 RAG？"
进入 Parallel：数据被复制两份，分别进入分支。
- 分支 A (context)：拿着问题去向量库查，查到文档，格式化。
  - 输出："RAG 是检索增强生成..."
- 分支 B (question)：RunnablePassthrough 是什么都不做，原样返回。
  - 输出："什么是 RAG？"
合并：Parallel 将两个分支的结果打包。
- 输出：{'context': 'RAG 是...', 'question': '什么是 RAG？'}

⚠️ 关键点：
RunnableParallel 的输出一定是一个字典。字典的 Key（如 context, question）由你在代码里定义的名字决定。

03 为什么需要 RunnablePassthrough？

很多初学者会问：“我直接在 Prompt 里用原始输入不行吗？”

不行。因为 RunnableParallel 要求它的每一个分支都必须是可运行对象（Runnable）。

vec_store.as_retriever() 是一个 Runnable。
但原始的 question 只是一个字符串，它不会“跑”，它只是数据。

为了让字符串也能在流水线里“占个坑”并传递下去，我们需要 RunnablePassthrough。它就像一个透明管道：

1	输入 "Hello" --> [Passthrough] --> 输出 "Hello"

它确保了 question 这个键值对能顺利进入最终的字典里，供后面的 Prompt 使用。

04 完整数据流追踪（Debug 视角）

让我们把整条链串起来，看数据是如何变形的。

chain = (
    RunnableParallel(...)  # 第 1 站
    | prompt               # 第 2 站
    | llm                  # 第 3 站
    | StrOutputParser()    # 第 4 站
)

第 1 站：RunnableParallel

输入："请总结简历" (String)
处理：检索文档 + 透传问题

输出：

{
  "context": "张三，5 年经验，擅长 Python...", 
  "question": "请总结简历"
}

第 2 站：Prompt (ChatPromptTemplate)

输入：上面的字典
处理：Prompt 模板里有 {context} 和 {question} 占位符，LangChain 会自动根据字典的 Key 进行填充。
输出：PromptValue (一段拼接好的完整文本，包含系统指令、上下文和问题)

第 3 站：LLM (DeepSeek)

输入：PromptValue
处理：大模型思考、生成。
输出：AIMessage (LangChain 的消息对象，包含文本内容)

第 4 站：StrOutputParser

输入：AIMessage
处理：提取其中的文本内容。
输出："张三拥有 5 年工作经验..." (String)

05 常见报错与避坑指南

结合之前的调试经验，这里有两个最容易踩的坑：

坑 1：类型不匹配 (TypeError)

报错：Expected a Runnable... Instead got an unsupported type: <class 'str'>
原因：在链式调用中，| 连接的东西必须是 Runnable 对象。

❌ 错误：| "一些字符串"
✅ 正确：| ChatPromptTemplate.from_template("一些字符串")
教训：Prompt 必须是模板对象，不能是纯字符串。

坑 2：字典 Key 对不上

报错：Prompt 输入缺少变量 {context}
原因：RunnableParallel 生成的字典 Key 必须与 Prompt 模板里的变量名完全一致。

Parallel 定义：context=...
Prompt 模板：{context}
教训：大小写敏感，拼写必须一致。

06 总结与记忆卡片

为了加深印象，请保存这张逻辑记忆卡：

组件	作用	输入	输出	比喻
**`	`**	连接符	上一步结果	下一步输入
`RunnablePassthrough`	透传	任意数据	原样数据	透明管道
`RunnableParallel`	并行打包	单份数据	字典 `{'key': val}`	分流合并器
`Prompt`	模板填充	字典	拼接后的文本	填空游戏
`OutputParser`	格式清洗	LLM 原始输出	纯净字符串	过滤器