LangChain 可组合链语法 LCEL 深度解析（LangChain Expression Language）

随着大语言模型（LLM）应用逐渐从 Demo 走向生产环境，工程复杂度显著提升。如何将 Prompt、模型、检索器、工具调用、结构化输出等组件进行可组合、可复用、可观测的编排，成为核心问题。

LangChain 在 2023 年提出并强化了一种新的链式表达范式 —— LCEL（LangChain Expression Language），即“LangChain 表达式语言”。它本质上是一种可组合的 Runnable 抽象层，用于声明式构建 LLM 工作流。

本文将系统解析：

1. LCEL 的设计目标
2. Runnable 抽象模型
3. 核心语法（| 管道、字典并行、映射）
4. 高阶用法（流式、批量、并发、结构化输出）
5. 与传统 Chain API 的差异
6. 工程最佳实践

---

一、LCEL 的设计目标

LCEL 的出现是为了解决传统 Chain API 的几个问题：

• 隐式执行逻辑（难调试）
• 嵌套结构复杂
• 可组合性弱
• 异步 / 流式支持不统一

LCEL 的核心理念：

一切皆 Runnable。

所有组件 —— Prompt、LLM、Retriever、OutputParser —— 都被抽象为实现统一接口的 Runnable 对象。

Runnable 支持：

• invoke() —— 单次调用
• batch() —— 批量
• stream() —— 流式
• ainvoke() —— 异步

这使得整个执行图具备函数式组合特征。

---

二、Runnable 抽象模型

LCEL 的类型签名可以理解为：

Runnable[Input, Output]

例如：

• PromptTemplate：dict -> str
• LLM：str -> str
• OutputParser：str -> structured data

当你用 | 连接时，本质是函数复合：

f | g == g(f(x))

---

三、LCEL 核心语法

1️⃣ 管道操作符 |

这是 LCEL 的核心语法糖。

from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI

prompt = ChatPromptTemplate.from_template("给我讲一个关于{topic}的笑话")
model = ChatOpenAI()

chain = prompt | model

chain.invoke({"topic": "程序员"})

执行流程：

dict → prompt → model → AIMessage

其可读性类似 Unix pipeline 或函数式编程。

---

2️⃣ 字典并行执行

LCEL 支持并行构造输入：

from langchain_core.runnables import RunnableParallel

chain = RunnableParallel({
    "joke": prompt | model,
    "analysis": another_prompt | model
})

输入一次，输出为：

{
  "joke": "...",
  "analysis": "..."
}

这在构建多视角分析时非常有用。

---

3️⃣ RunnablePassthrough（透传）

在构造复杂输入时，经常需要保留原始输入。

from langchain_core.runnables import RunnablePassthrough

chain = {
    "context": retriever,
    "question": RunnablePassthrough()
} | prompt | model

等价于：

input → retriever 得 context
input → passthrough 保留 question
→ 合并成 dict → prompt

这是实现 RAG 的关键模式。

---

四、构建一个完整 RAG LCEL 示例

RAG 结构：

用户问题 → 检索 → Prompt → LLM → 输出

rag_chain = (
    {
        "context": retriever,
        "question": RunnablePassthrough()
    }
    | prompt
    | model
    | StrOutputParser()
)

这里每一步都是 Runnable。

与旧版 RetrievalQAChain 相比：

• 更显式
• 更灵活
• 更易插拔

---

五、高阶能力

1️⃣ 批量执行

chain.batch([
    {"topic": "AI"},
    {"topic": "数学"}
])

自动并发调度。

---

2️⃣ 流式输出

for chunk in chain.stream({"topic": "哲学"}):
    print(chunk)

LCEL 原生支持 streaming，无需额外封装。

---

3️⃣ 结构化输出

结合 Pydantic：

from pydantic import BaseModel

class Joke(BaseModel):
    setup: str
    punchline: str

structured_chain = prompt | model.with_structured_output(Joke)

模型输出自动校验。

---

4️⃣ 条件分支（Router）

可以通过 RunnableLambda 实现动态分支逻辑：

from langchain_core.runnables import RunnableLambda

def route(x):
    if "代码" in x:
        return code_chain
    return normal_chain

router = RunnableLambda(route)

---

六、LCEL vs 旧版 Chain API

维度	旧 Chain API	LCEL
组合方式	嵌套对象	函数式管道
可读性	中等	高
流式支持	不统一	原生
并行	较弱	原生支持
可观测性	依赖外部工具	更透明

LCEL 本质是将 LangChain 变成一个声明式数据流图构建框架。

---

七、工程实践建议

1️⃣ 始终保持“单一职责 Runnable”

不要把太多逻辑塞进一个 RunnableLambda。

2️⃣ 显式使用 OutputParser

避免直接依赖模型自由文本。

3️⃣ 使用 LangSmith 进行可观测

配合 LangSmith 可追踪每个 Runnable 的执行图。

4️⃣ 生产环境建议

• 配合缓存（Redis / SQLite）
• 明确超时与重试策略
• 控制并发上限
• 加入日志与指标

---

八、LCEL 的本质理解

LCEL 并不是“语法糖”，它是：

• Runnable 抽象统一化
• 数据流声明式表达
• 可组合函数式编排
• 执行模型标准化

可以将其类比为：

• 深度学习中的计算图
• Unix pipeline
• Rx 流式编程

---

总结

LCEL 代表了 LangChain 向工程化和可组合架构的一次关键升级。

如果你正在构建：

• RAG 系统
• Agent
• 多模型协作系统
• 工具调用链
• 复杂 Prompt 编排系统

LCEL 几乎是必选方案。

它让 LLM 应用开发从“脚本式拼接”进化为“数据流编排”。

示例：LangChain字典映射节点

代码

chain = (
    {
        "input": RunnablePassthrough(),
        "context": retriever | format_document
    }
    | self.prompt_template
    | print_prompt
    | self.chat_model
    | StrOutputParser()
)

解释

这几行是在用 LangChain 的可组合链语法（LCEL）构造一个「输入映射步骤」，含义是：

{
    "input": RunnablePassthrough(),
    "context": retriever | format_document
}

逐条解释：

• 整个大括号 { ... }：
  表示一个「字典映射节点」，把上游传进来的同一份输入，分别加工成多个字段，传给后面的链。

  • 这里会生成一个形如 {"input": xxx, "context": yyy} 的字典，给下游的 self.prompt_template 使用。

• "input": RunnablePassthrough()：

  • RunnablePassthrough() 的意思是「原样透传上游输入」，不做任何处理。
  • 所以最后字典里的 input 字段就是：用户原始问题。

• "context": retriever | format_document：

  • retriever：向量检索器，接收用户问题，返回若干 Document。
  • |：LCEL 的「管道」操作符，表示前一个的输出作为后一个的输入。
  • format_document：把检索到的 Document 列表拼成一段字符串（你定义的「文档片段 + 元数据」）。
  • 所以 context 字段就是：根据用户问题检索到的文档内容字符串。

总结一下：
这 55–58 行的作用是：对同一个用户输入，同时生成两个键值对——input（原问题）和 context（检索出的参考资料文本），然后再一起喂给后面的 ChatPromptTemplate。

→问AI：python有个缺点就是变量太随意了，让人搞不太清楚这个input、context被谁处理了，要一个个点进去看

←AI回复

确实，你说的是 Python（尤其配合 LangChain 这类“链式组合”）的一个现实问题：名字随意 + 类型不显式，一眼很难看出 input、context 在哪一层被谁消费。

简单给你几个“缓解痛点”的思路，方便你自己读/写这种代码时不至于迷路：

1. 把“数据结构”写在注释里

比如你现在这段链，可以直接在 _get_chain 里写清楚“这一坨到底变成什么”：

# 这里的整体输出是一个 dict:
# {
#   "input": <用户原始问题字符串>,
#   "context": <根据问题检索出的文档拼接字符串>
# }
chain_inputs = {
    "input": RunnablePassthrough(),
    "context": retriever | format_document,
}

后面再写：

# prompt_template 期望的输入字段就是上面的 "input" 和 "context"
chain = chain_inputs | self.prompt_template | ...

这样你以后回来看，不用点进每一个 callable，只看注释就知道字段流向。

2. 显式写类型（哪怕只是注释 / typing）

给关键函数加上类型标注，哪怕只是注释级别的，也会清晰很多，比如：

from typing import Dict, Any

def _get_chain(self):
    retriever = self.vector_service.get_retriever()

    def format_document(docs: list[Document]) -> str:
        ...

    chain: Runnable[Dict[str, Any], str] = (
        {
            "input": RunnablePassthrough(),          # str -> str
            "context": retriever | format_document   # str -> str
        }
        | self.prompt_template                      # Dict[str, str] -> ChatPromptValue
        | ...
    )
    return chain

看到 Runnable[Dict[str, Any], str]，你就知道：这个链的输入是一个 dict，输出是 str。

3. 对“中间结果”起名字，而不是一坨连到底

比如把中间步骤拆开命名：

inputs_mapping = {
    "input": RunnablePassthrough(),
    "context": retriever | format_document,
}

rag_chain = inputs_mapping | self.prompt_template | print_prompt | self.chat_model | StrOutputParser()

这样你可以一句话理解：inputs_mapping 负责构造 {"input": ..., "context": ...}，rag_chain 是最终链。

4. 约定字段命名：见名知意

你现在的 input / context 可以考虑稍微“长一点但不含糊”的命名，比如：

• user_input 代替 input
• retrieved_context 代替 context

只改字段名，不改逻辑，阅读体验会好一大截。

---

总结：
Python / LangChain 的确容易造成“变量太随意、数据流看不清”的感觉，但你完全可以通过 类型标注 + 注释写清中间结构 + 给中间步骤起名字 + 用自解释的字段名，把“到底谁处理了 input/context”这件事写死在代码上，这样以后几乎不用点进去也能看懂。