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

引言在上一篇文章中我们讨论了链式调用的通用原理——从return this到流式 API 的设计哲学。在 AI 应用开发领域链式调用正以一种全新的形态重新定义开发者体验LangChain Expression LanguageLCEL。不同于传统的对象方法链LCEL 使用管道运算符|将多个独立的 Runnable 组件串联成一个可执行的“链”Chain每个组件接收前一个的输出并产生新的输出整个过程呈现出声明式、可组合、可流式的鲜明特征。本文将从 PDF 笔记中提炼的 LangChain Chains 实践出发深入剖析 LCEL 的链式机制并通过代码示例展示其如何简化大模型应用开发同时结合工程经验讨论其优势与陷阱。一、链式调用基础回顾两种范式在进入 LCEL 之前有必要简要回顾链式调用的两种经典实现范式详见前文范式实现方式代表案例特点可变链式方法返回thisjQuery、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 响应AIMessageoutput_parser.invoke(响应)→ 解析为结构化输出如字符串、JSON每个组件都是无状态的不保存任何内部状态除显式配置的记忆模块。这种设计使链式组合变得极其灵活——你可以轻松地插入、替换或重用任意组件。2.3 与经典链式调用的差异特性经典对象链LCEL 管道链链接方式方法调用.管道运算符|返回值this或新对象新的 Runnable 组合体状态管理实例内部可变状态组件无状态状态由外部管理组合粒度方法级组件级Prompt、Model、Parser 等扩展性需修改类定义通过组合任意 RunnableLCEL 将链式调用从“方法串联”提升为“组件流水线”更符合函数式编程的管道思维。三、代码示例从基础链到带记忆的对话链以下代码均基于 PDF 笔记中的实践使用 LangChain 与本地 Ollama 或云模型。3.1 基础 LCEL 链Prompt Model Parserimport 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(verboseTrue, overrideTrue) # 初始化 LLM使用 Qwen 或 Ollama llm ChatOpenAI( modelqwen3.7-max, base_urlos.getenv(QWEN_BASE_URL), api_keyos.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, flushTrue)每个中间组件都支持流式传递实现“逐词生成”体验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_namehistory), # ← 历史消息将插入此处 (human, {input}) ]) # 基础链 llm ChatOpenAI(model_nameqwen3.7-max, ...) chain prompt | llm | StrOutputParser() # 包装为带历史记录的链 chain_with_history RunnableWithMessageHistory( chain, get_session_history, input_messages_keyinput, # 指定输入字段名 history_messages_keyhistory # 指定历史字段名 ) # 第一次调用 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 强大组合能力的基石。