当前位置：首页 > news >正文

20250913-02: Langchain概念：表达式语言（LCEL）

news 2025/9/14 7:21:13

20250913-02: Langchain概念：表达式语言（LCEL）

任务

🎯 学习目标

🔗 核心概念

LangChain 表达式语言（LCEL）

我应该使用 LCEL 吗？

组合原语

RunnableSequence：顺序调用
RunnableParallel：并行调用

组合语法：简写

运算符

.pipe 方法

强制转换

字典到 RunnableParallel
函数到 RunnableLambda

传统链

复习知识点

LangChain表达式语言 (LCEL) - 核心概念
LCEL 的优势 (关键说明)
LCEL 适用场景 (重要指南)
组合原语
组合语法 (关键语法糖)
与传统链的对比

脚注

关键：“声明式” vs “命令式”

任务

阅读 LCEL 官方文档
理解 | 操作符的底层是 RunnableSequence

‍

🎯 学习目标

理解 LangChain 的“运行时引擎”——所有组件如何被统一调度和编排。

🔗 核心概念

可运行接口（Runnable）
LangChain 表达式语言（LCEL）
回调（Callbacks）
追踪（Tracing）
流式传输（Streaming）
异步编程（Async）

LangChain 表达式语言（LCEL）

The LangChain Expression Language (LCEL) 采用声明式方法，从现有 Runnables 构建新的Runnables。

这意味着您描述的是应该发生什么，而不是如何发生，从而允许 LangChain 优化链的运行时执行。

**LCEL 是一种让你通过像拼管道一样的方式（** | ），来声明式地组合AI应用各个模块的语言。你只需要描述数据“应该”从哪里流到哪里，而无需操心具体“如何”实现每一步的调用和传递。这种抽象使得 LangChain 框架能够在背后智能地优化整个链路的执行效率、提供重试和流式传输等高级功能，让开发者更专注于逻辑本身而非实现细节。

简单来说，LCEL 让构建AI应用变得更简单、更强大、更高效。

LangChain 以多种方式优化使用 LCEL 构建的链的运行时执行

** 优化并行执行 **：使用 RunnableParallel 并行运行 Runnables，或者使用 Runnable 批处理 API 并行地通过给定链运行多个输入。并行执行可以显著减少延迟，因为处理可以并行完成而不是按顺序进行。
** 保证异步支持 **：任何使用 LCEL 构建的链都可以使用 Runnable 异步 API 异步运行。这在服务器环境中运行链时非常有用，您希望并发处理大量请求。
** 简化流式传输 **：LCEL 链可以进行流式传输，从而在链执行时允许增量输出。LangChain 可以优化输出的流式传输，以最大限度地减少首个令牌时间（从聊天模型或大型语言模型输出的第一个块出现所需的时间）。

‍

其他优势包括

无缝 LangSmith 追踪 随着您的链变得越来越复杂，了解每个步骤中到底发生了什么变得越来越重要。使用 LCEL，所有步骤都会自动记录到 LangSmith，以实现最大程度的可观察性和可调试性。
标准 API：由于所有链都使用 Runnable 接口构建，因此它们可以像任何其他 Runnable 一样使用。
可与 LangServe 部署：使用 LCEL 构建的链可以部署用于生产环境。

‍

我应该使用 LCEL 吗？

LCEL 是一个编排解决方案——它允许 LangChain 以优化的方式处理链的运行时执行。

以下是一些指南

如果您只进行一次 LLM 调用，则不需要 LCEL；而是直接调用底层的聊天模型。
如果您有一个简单的链（例如，提示 + llm + 解析器，简单的检索设置等），如果您正在利用 LCEL 的优势，那么 LCEL 是一个合理的选择。
如果您正在构建一个复杂的链（例如，具有分支、循环、多个代理等），请改用LangGraph。请记住，您始终可以在 LangGraph 中的单个节点内使用 LCEL。

组合原语

LCEL 链是通过将现有 Runnables 组合在一起构建的。两个主要的组合原语是 **== ** RunnableSequence ** 和 ** RunnableParallel==。

您可以在LangChain Core API 参考中找到所有组合原语的列表。

RunnableSequence：顺序调用

RunnableSequence 是一个组合原语，它允许您按顺序 “链接”多个 runnable，其中一个 runnable 的输出作为下一个 runnable 的输入。

from langchain_core.runnables import RunnableSequence
chain = RunnableSequence([runnable1, runnable2])final_output = chain.invoke(some_input)
# ==> 如下内容顺序调用
output1 = runnable1.invoke(some_input)
final_output = runnable2.invoke(output1)

‍

RunnableParallel：并行调用

runnableParallel 是一个组合原语，它允许您并发运行多个 runnable，并为每个 runnable 提供相同的输入。

将产生一个 final_output 字典，其键与输入字典相同，但值替换为相应 runnable 的输出。

from langchain_core.runnables import RunnableParallel
chain = RunnableParallel({"key1": runnable1,"key2": runnable2,
})final_output = chain.invoke(some_input)
# ==> 如下并行
{"key1": runnable1.invoke(some_input),"key2": runnable2.invoke(some_input),
}

请记住，runnable 是并行执行的，因此虽然结果与上面所示的字典推导式相同，但执行时间要快得多

‍

提醒

RunnableParallel支持同步和异步执行（所有 Runnables 都支持）。

对于同步执行，RunnableParallel 使用 ThreadPoolExecutor 并发运行 runnable。
对于异步执行，RunnableParallel 使用 asyncio.gather 并发运行 runnable。

组合语法：简写

RunnableSequence 和 RunnableParallel 的用法非常常见，因此我们创建了它们的简写语法。这有助于使代码更具可读性和简洁性

运算符

我们重载了 | 运算符，以从两个 Runnables 创建一个 RunnableSequence。

chain = runnable1 | runnable2   
# ===》 
chain = RunnableSequence([runnable1, runnable2])

`.pipe` 方法

如果您对运算符重载有道德上的顾虑，您可以改用 .pipe 方法。这等同于 | 运算符。

chain = runnable1.pipe(runnable2)

强制转换

LCEL 应用 自动类型强制转换 ，以简化链的组合。

如果您不理解类型强制转换，您可以直接使用 RunnableSequence 和 RunnableParallel 类。

字典到 RunnableParallel

在 LCEL 表达式中，字典会自动转换为 RunnableParallel。

mapping = {"key1": runnable1,"key2": runnable2,
}chain = mapping | runnable3
# ==> 自动转换如下
chain = RunnableSequence([RunnableParallel(mapping), runnable3])

‍

注意

您必须小心，因为 mapping 字典不是一个 RunnableParallel 对象，它只是一个字典。这意味着以下代码将引发 AttributeError
mapping.invoke(some_input)

‍

函数到 RunnableLambda

在 LCEL 表达式中，函数会自动转换为 RunnableLambda。

def some_func(x):return xchain = some_func | runnable1
# ==》 函数自动转换
chain = RunnableSequence([RunnableLambda(some_func), runnable1])

‍

注意

您必须小心，因为 lambda 函数不是一个 RunnableLambda 对象，它只是一个函数。这意味着以下代码将引发 AttributeError
lambda x: x + 1.invoke(some_input)

传统链

LCEL 旨在提供行为和自定义方面的连贯性，以取代旧的子类化链，例如 LLMChain 和 ConversationalRetrievalChain。

许多这些传统链隐藏了诸如提示之类的关键细节，并且随着各种可行模型的出现，自定义变得越来越重要。

如果您目前正在使用这些传统链之一，请参阅本指南以获取迁移指导。

复习知识点

LangChain表达式语言 (LCEL) - 核心概念

LCEL 定义: 一种采用声明式方法来组合 Runnable 构建新 Runnable（链）的语言。
核心思想: 你描述应该发生什么，而不是如何发生，将执行优化交给 LangChain。
链的本质: 使用 LCEL 创建的 Runnable 称为“链”，它实现了完整的 Runnable 接口。

LCEL 的优势 (关键说明)

并行执行: 使用 RunnableParallel 或批处理 API 并行运行任务，显著减少延迟。
异步支持: 任何 LCEL 链都天然支持异步调用，利于高并发服务器环境。
简化流式传输: LCEL 链可以进行增量输出（逐词输出） ，并优化以减少首个令牌时间。
无缝集成 LangSmith: 所有步骤自动记录，提供极佳的可观察性和可调试性。
标准 API: 所有链都基于 Runnable 接口，拥有统一的使用方式（invoke, batch, stream, ainvoke）。
易于部署: 可以使用 LangServe 直接部署用于生产环境。

LCEL 适用场景 (重要指南)

简单调用: 如果只需一次 LLM 调用，不需要使用 LCEL，直接调用模型即可。
简单链: 对于提示+LLM+解析器或简单检索链，LCEL 是合理的选择。
复杂链: 对于需要复杂状态、分支、循环或多代理的应用，应使用 LangGraph，但可在其节点内使用 LCEL。

组合原语

两大核心原语: RunnableSequence（顺序执行）和 RunnableParallel（并行执行）。
RunnableSequence: 按顺序链接多个 Runnable，前一个的输出作为后一个的输入。
RunnableParallel: 并发运行多个 Runnable，并为每个提供相同的输入，输出一个字典。

组合语法 (关键语法糖)

**管道运算符 (** | ) : runnableA | runnableB 等价于创建 RunnableSequence([runnableA, runnableB])。
.pipe() 方法: runnableA.pipe(runnableB) 是 | 运算符的等效替代。
自动类型转换: LCEL 会自动进行类型强制转换以简化代码。
字典转 RunnableParallel: 在管道中，字典 {"key": runnable} 会自动转换为 RunnableParallel。
- 关键说明: 字典本身不是 Runnable，直接调用 .invoke() 会报错。
函数转 RunnableLambda: 在管道中，函数会自动转换为 RunnableLambda。
- 关键说明: 函数本身不是 Runnable，直接调用 .invoke() 会报错。

与传统链的对比

设计目标: LCEL 旨在提供更一致的行为和更强的自定义能力，以取代旧的基于子类化的链（如 LLMChain）。
迁移建议: 旧链通常隐藏实现细节，LCEL 则更加透明和灵活，官方提供了从传统链迁移的指南。

脚注

关键：“声明式” vs “命令式”

这是理解这段话的精髓。我们用一个简单的类比来解释：

任务：做一杯咖啡

命令式（Imperative）方法（描述“如何”做）：
1. 走到咖啡机前。
2. 拿出一个滤杯，放入滤纸。
3. 用量勺取15克咖啡粉倒入滤杯。
4. 用热水壶烧水至92摄氏度。
5. 缓慢注水50毫升，等待30秒进行“闷蒸”。
6. 继续缓慢注水至250毫升。
7. 将冲好的咖啡倒入杯子。
这种方法需要你详细规定每一个步骤和具体操作。
声明式（Declarative）方法（描述“应该”发生什么）：
“我要一杯手冲咖啡。”
你只描述了最终的目标和结果，而没有规定具体的操作细节。咖啡师（在这里相当于 LangChain）会根据你的要求，自行决定最佳的步骤、水温、水流等来优化制作过程。