LangChain 表达式语言 (LCEL)
The LangChain Expression Language (LCEL) 采用声明式方法,基于现有的 Runnables 构建新的 Runnables。
这意味着你描述的是应该发生什么,而不是如何发生,从而允许 LangChain 优化链的运行时执行。
我们通常将使用 LCEL 创建的 Runnable 称为“链”。重要的是要记住,“链”Runnable,并且它实现了完整的 可运行接口。
- The LCEL 速查表 展示了涉及 Runnable 接口和 LCEL 表达式的常见模式。
- 请参阅以下关于使用 LCEL 执行常见任务的操操作指南列表。
- 一组内置的
Runnables可以在 LangChain Core API 参考 中找到。许多这些可运行对象在 LangChain 中使用 LCEL 组合自定义“链”时非常有用。
LCEL 的优势
LangChain 通过多种方式优化了使用 LCEL 构建的链在运行时的执行效率:
- 优化的并行执行:使用 RunnableParallel 并行运行 Runnable,或使用 Runnable Batch API 并行地将多个输入通过给定链。并行执行可以显著降低延迟,因为处理可以并行进行而非顺序进行。
- 保证异步支持:任何使用 LCEL 构建的链都可以通过 Runnable Async API 以异步方式运行。这在服务器环境中非常有用,特别是在需要并发处理大量请求时。
- 简化流式处理:LCEL 链支持流式传输,允许在链执行过程中逐步输出结果。LangChain 可优化输出流式处理,以最小化首令牌时间(即从 聊天模型 或 大语言模型 输出的第一个数据块出现所经过的时间)。
其他好处包括:
- 无缝的 LangSmith 追踪 随着您的链变得越来越复杂,理解每一步具体发生了什么变得日益重要。 使用 LCEL,所有步骤都会自动记录到 LangSmith,以实现最佳的可见性和可调试性。
- 标准 API: 由于所有链都是使用 Runnable 接口构建的,因此它们可以像任何其他 Runnable 一样使用。
- 可部署于 LangServe: 使用 LCEL 构建的链可以使用为生产用途进行部署。
我应该使用 LCEL 吗?
LCEL 是一个编排解决方案——它允许 LangChain 以优化的方式处理链的运行时执行。
虽然我们已看到用户在生产环境中运行包含数百个步骤的链,但我们通常建议使用 LCEL 来处理更简单的编排任务。当应用需要复杂的状态管理、分支、循环或多个智能体时,我们建议用户利用 LangGraph。
在 LangGraph 中,用户定义指定应用程序流程的图。这使得用户在需要时可以在各个节点内继续使用 LCEL,同时能够轻松定义更易于阅读和维护的复杂编排逻辑。
以下是一些指南:
- 如果您只进行一次 LLM 调用,则不需要使用 LCEL;相反,请直接调用底层的聊天模型。
- 如果您有一个简单的链(例如:提示词 + LLM + 解析器,简单的检索设置等),并且您正在利用 LCEL 的优势,那么 LCEL 是一个合理的选择。
- 如果您正在构建复杂的链(例如,包含分支、循环、多个智能体等),请使用 LangGraph。请记住,您始终可以在 LangGraph 的各个节点中使用 LCEL。
组合原语
LCEL 条链是通过组合现有的 Runnables 构建而成的。两个主要的组合原语是 RunnableSequence 和 RunnableParallel。
许多其他组合原语(例如,RunnableAssign)可以被视为这两个原语的变体。
您可以在 LangChain Core API 参考 中找到所有组合原语的列表。
RunnableSequence
RunnableSequence 是一个组合原语,允许您按顺序“链式”调用多个可运行对象,其中一个可运行对象的输出作为下一个可运行对象的输入。
from langchain_core.runnables import RunnableSequence
chain = RunnableSequence([runnable1, runnable2])
调用 chain 并传入一些输入:
final_output = chain.invoke(some_input)
对应以下内容:
output1 = runnable1.invoke(some_input)
final_output = runnable2.invoke(output1)
runnable1 和 runnable2 是占位符,代表您希望链接在一起的任意 Runnable。
RunnableParallel
RunnableParallel 是一个组合原语,允许您并发运行多个可运行对象,并向每个对象提供相同的输入。
from langchain_core.runnables import RunnableParallel
chain = RunnableParallel({
"key1": runnable1,
"key2": runnable2,
})
调用 chain 并传入一些输入:
final_output = chain.invoke(some_input)
将生成一个 final_output 字典,其键与输入字典相同,但值被替换为相应可运行对象的输出。
{
"key1": runnable1.invoke(some_input),
"key2": runnable2.invoke(some_input),
}
回忆一下,可运行对象是并行执行的,因此虽然结果与上述字典推导式相同,但执行时间要快得多。
RunnableParallel支持同步和异步执行(所有Runnables均如此)。
- 对于同步执行,
RunnableParallel使用 线程池执行器(ThreadPoolExecutor) 并发运行可运行对象。 - 对于异步执行,
RunnableParallel使用 asyncio.gather 并发运行可运行对象。
组合语法
RunnableSequence 和 RunnableParallel 的使用非常普遍,因此我们创建了一种简写语法来使用它们。这有助于使代码更易读且更简洁。
The | 操作符
我们已重载 运算符,以便使用两个 Runnables 创建一个 RunnableSequence。
chain = runnable1 | runnable2
等同于:
chain = RunnableSequence([runnable1, runnable2])
The .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 x
chain = some_func | runnable1
它会自动转换为以下内容:
chain = RunnableSequence([RunnableLambda(some_func), runnable1])
你必须小心,因为 lambda 函数不是 RunnableLambda 对象,它只是一个函数。这意味着以下代码将引发一个 AttributeError:
lambda x: x + 1.invoke(some_input)
遗留链
LCEL 旨在提供一致的行为和定制能力,以取代基于遗留子类的链(如 LLMChain 和 ConversationalRetrievalChain)。许多这些遗留链隐藏了重要细节(例如提示词),而随着越来越多可行的模型出现,定制化变得越来越重要。
如果您目前正在使用这些遗留链,请参阅本指南以获取迁移指导。
有关如何使用 LCEL 执行特定任务的指南,请查看相关操操作指南。