LangChain LCEL 链式调用：从管道运算符到可组合的 AI 应用

pann_zxw

306人浏览 · 2026-07-04 16:36:33

pann_zxw · 2026-07-04 16:36:33 发布

引言

在上一篇文章中，我们讨论了链式调用的通用原理——从 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，自动检索相关文档作为 context