LangChain LCEL 链式调用：从管道运算符到可组合的 AI 应用
引言
在上一篇文章中，我们讨论了链式调用的通用原理——从return this到流式 API 的设计哲学。在 AI 应用开发领域，链式调用正以一种全新的形态重新定义开发者体验：LangChain Expression Language（LCEL）。不同于传统的对象方法链，LCEL 使用管道运算符|将多个独立的 Runnable 组件串联成一个可执行的“链”（Chain），每个组件接收前一个的输出并产生新的输出，整个过程呈现出声明式、可组合、可流式的鲜明特征。
本文将从 PDF 笔记中提炼的 LangChain Chains 实践出发，深入剖析 LCEL 的链式机制，并通过代码示例展示其如何简化大模型应用开发，同时结合工程经验讨论其优势与陷阱。
一、链式调用基础回顾：两种范式
在进入 LCEL 之前，有必要简要回顾链式调用的两种经典实现范式（详见前文）：
| 范式 | 实现方式 | 代表案例 | 特点 |
|---|---|---|---|
| 可变链式 | 方法返回this | jQuery、Builder 模式 | 操作同一对象，状态可变 |
| 不可变链式 | 方法返回新对象 | Promise、Java Stream | 每次产生新实例，无副作用 |
LCEL 则走出了一条第三条道路：它不直接依赖方法返回对象或新实例，而是通过运算符重载（Python 的__or__）将多个独立组件组合成一个新的Runnable 对象。这个组合体本身也是 Runnable，从而支持无限嵌套。这种设计更接近于函数组合（Function Composition），而非传统的对象链。
二、LCEL 核心原理：管道运算符与 Runnable 协议
2.1 Runnable 统一接口
LangChain 将所有可执行单元抽象为Runnable协议，每个 Runnable 都实现invoke、stream、batch等方法。无论是 Prompt 模板、LLM 模型还是输出解析器，都遵循这一接口。
# 任何 Runnable 都具有统一调用方式 runnable.invoke(input) # 同步调用 runnable.stream(input) # 流式输出 runnable.batch([inputs]) # 批量处理2.2 管道运算符|的组合语义
LCEL 的核心创新在于使用|运算符将两个 Runnable 组合成一个新的 Runnable，其语义为：前一个 Runnable 的输出成为后一个 Runnable 的输入。
chain = prompt | model | output_parser这行代码等价于：
# 手动嵌套调用 chain = RunnableSequence(prompt, model, output_parser)当调用chain.invoke({"topic": "编程"})时，执行流程如下：
prompt.invoke({"topic": "编程"})→ 生成完整的消息列表model.invoke(消息列表)→ 返回 LLM 响应（AIMessage）output_parser.invoke(响应)→ 解析为结构化输出（如字符串、JSON）
每个组件都是无状态的，不保存任何内部状态（除显式配置的记忆模块）。这种设计使链式组合变得极其灵活——你可以轻松地插入、替换或重用任意组件。
2.3 与经典链式调用的差异
| 特性 | 经典对象链 | LCEL 管道链 |
|---|---|---|
| 链接方式 | 方法调用. | 管道运算符| |
| 返回值 | this或新对象 | 新的 Runnable 组合体 |
| 状态管理 | 实例内部可变状态 | 组件无状态，状态由外部管理 |
| 组合粒度 | 方法级 | 组件级（Prompt、Model、Parser 等） |
| 扩展性 | 需修改类定义 | 通过组合任意 Runnable |
LCEL 将链式调用从“方法串联”提升为“组件流水线”，更符合函数式编程的管道思维。
三、代码示例：从基础链到带记忆的对话链
以下代码均基于 PDF 笔记中的实践，使用 LangChain 与本地 Ollama 或云模型。
3.1 基础 LCEL 链：Prompt + Model + Parser
import os from dotenv import load_dotenv from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser from langchain_openai import ChatOpenAI load_dotenv(verbose=True, override=True) # 初始化 LLM（使用 Qwen 或 Ollama） llm = ChatOpenAI( model="qwen3.7-max", base_url=os.getenv("QWEN_BASE_URL"), api_key=os.getenv("QWEN_API_KEY") ) # 定义提示模板 prompt = ChatPromptTemplate.from_messages([ ("system", "你是一位资深的{role}专家，请用简洁的语言回答。"), ("human", "请解释什么是{topic}？") ]) # 输出解析器 output_parser = StrOutputParser() # 构建 LCEL 链 chain = prompt | llm | output_parser # 调用链 result = chain.invoke({"role": "Python", "topic": "装饰器"}) print(result) # 输出：装饰器是 Python 中用于修改函数行为的高级函数...执行流程解析：
prompt.invoke()根据输入变量生成消息列表。llm.invoke()获取模型响应（AIMessage）。output_parser.invoke()提取content字段返回字符串。
3.2 流式输出链
LCEL 天然支持流式，只需使用stream()方法：
for chunk in chain.stream({"role": "诗人", "topic": "春天"}): print(chunk.content, end="", flush=True)每个中间组件都支持流式传递，实现“逐词生成”体验，PDF 中也有类似示例（使用llm.stream）。
3.3 带会话记忆的链：RunnableWithMessageHistory
在对话应用中，需要记住历史消息。LangChain 提供了RunnableWithMessageHistory包装器，它会自动将历史消息注入到链中。
from langchain_core.chat_history import InMemoryChatMessageHistory from langchain_core.runnables import RunnableWithMessageHistory from langchain_core.prompts import MessagesPlaceholder # 存储各会话的历史记录 store = {} def get_session_history(session_id: str): if session_id not in store: store[session_id] = InMemoryChatMessageHistory() return store[session_id] # 定义提示模板，包含历史消息占位符 prompt = ChatPromptTemplate.from_messages([ ("system", "你是AI助手"), MessagesPlaceholder(variable_name="history"), # ← 历史消息将插入此处 ("human", "{input}") ]) # 基础链 llm = ChatOpenAI(model_name="qwen3.7-max", ...) chain = prompt | llm | StrOutputParser() # 包装为带历史记录的链 chain_with_history = RunnableWithMessageHistory( chain, get_session_history, input_messages_key="input", # 指定输入字段名 history_messages_key="history" # 指定历史字段名 ) # 第一次调用 response1 = chain_with_history.invoke( {"input": "我叫张三"}, config={"configurable": {"session_id": "user123"}} ) print(response1) # 你好，张三！ # 第二次调用（自动携带历史） response2 = chain_with_history.invoke( {"input": "我刚才说我叫什么名字？"}, config={"configurable": {"session_id": "user123"}} ) print(response2) # 你说你叫张三。这里RunnableWithMessageHistory实际上是一个装饰器模式的实现，它将原始链包装，在每次invoke前从存储中读取历史并注入，调用后将新的消息追加到历史中，从而实现有记忆的对话。
四、LCEL 链式调用的优缺点
4.1 优势
1. 声明式构建，可读性强
LCEL 使用|清晰地表达了数据流方向，代码即文档。开发者可以一眼看出整个处理流程：用户输入 → Prompt 模板化 → 模型推理 → 输出解析。
2. 高度可组合
任意两个 Runnable 都可以组合，组合后的产物仍是 Runnable，因此可以无限嵌套。这种设计使得重用和测试变得极为容易——你可以将复杂链拆解为多个小链，分别测试后再组装。
3. 内置流式与异步支持
所有 LCEL 链都天然支持stream、batch和ainvoke，无需额外适配。这对于构建实时聊天应用至关重要。
4. 便捷的中间件扩展
通过RunnablePassthrough、RunnableLambda等工具，可以在链中插入自定义逻辑（日志、格式化、条件分支等），增强灵活性。
4.2 劣势
1. 调试困难（管道中的错误定位）
当一条长链出错时，错误堆栈往往指向链的invoke入口，难以快速定位是哪个组件出错。虽然 LangChain 提供了with_config和with_listeners辅助调试，但相比传统逐行调试仍有不足。
2. 隐式类型转换带来的困惑
组件之间的类型约束是隐式的——prompt输出消息列表，model输入消息列表，parser输入 AIMessage。如果组合顺序不当（例如将 parser 放在 model 前面），会抛出运行时错误，缺乏编译时检查。
3. 性能开销
每个组件的invoke都存在函数调用和序列化开销，对于极其简单的场景，LCEL 可能比直接调用 LLM 更慢。
4. 状态管理的复杂性
虽然RunnableWithMessageHistory提供了便捷的记忆支持，但其内部状态存储（示例中的store字典）需要自行管理持久化和并发安全，在生产环境中需要进一步封装。
五、实际应用场景
5.1 RAG（检索增强生成）管道
RAG 是 LCEL 的典型应用场景，链式组合检索器与生成器：
retriever = vectorstore.as_retriever() rag_chain = ( {"context": retriever, "question": RunnablePassthrough()} | prompt | llm | parser ) # 调用时，输入 question，自动检索相关文档作为 context5.2 多步骤 Agent 工作流
LCEL 可以组合多个工具调用和决策逻辑，构建自主 Agent。虽然复杂 Agent 通常需要AgentExecutor，但 LCEL 可用于构建子链，如“思考 → 工具调用 → 总结”。
5.3 对话式助手（带记忆）
如第三节所示，RunnableWithMessageHistory使开发者能够以极少的代码实现多轮对话管理。配合 Redis 或数据库存储，即可实现跨请求的持久化记忆。
5.4 流式响应接口
在 Web 应用中，使用 LCEL 的stream()可以实现打字机效果，提升用户体验。FastAPI 结合StreamingResponse可轻松集成。
六、设计启示：链式调用的新范式
LangChain LCEL 重新定义了链式调用在 AI 工程中的角色——它不再是简单的“方法串联”，而是面向数据流的组件编排语言。这种设计汲取了函数式编程（管道操作）、响应式编程（流式）和面向对象（Runnable 协议）的精华，为开发者提供了一种统一、可扩展的构建体验。
从技术层面看，LCEL 的成功离不开 Python 的运算符重载能力，但其背后更深层的设计原则是“单一职责”与“组合优于继承”——每个 Runnable 只做一件事，但通过管道可以组合出无限复杂的行为。
对于开发者而言，理解 LCEL 的链式机制不仅是使用 LangChain 的前提，更是构建可维护、可测试 AI 应用的关键。“Chain 本身也是 Runnable，可以继续调用”——这条简洁的规则，正是 LCEL 强大组合能力的基石。