作者
简而言之
在不到一年的时间里，LangChain已成为与大型语言模型（LLMs）交互最常用的Python 库之一，但 LangChain 主要还是一个用于概念验证（POC）的库，因为它缺乏构建复杂且可扩展应用程序的能力。
2023年8月，随着LangChain 发布LangChain 表达式语言（LCEL）——一种弥合概念验证（POC）与生产环境之间差距的新语法，一切发生了改变。本文将带您深入了解 LCEL 的方方面面，向您展示它如何简化自定义链的创建，以及如果您正在构建LLM 应用程序，为何必须学习它！
提示词、大型语言模型和链式生成，让我们温习一下
在深入探讨 LCEL 语法之前，我认为先回顾一下 LangChain 中的概念（例如 LLM、Prompt 甚至 Chain）会很有帮助。
LLM：在 LangChain 中，LLM 是围绕用于生成内容（如 OpenAI GPT-3.5、Claude 等）的模型所建立的抽象概念。
提示：这是 LLM 对象的输入，它将向 LLM 提出问题并指定其目标。
链：指对大型语言模型（LLM）的一系列调用，或任何数据处理步骤。
既然定义已经说清楚了，那我们就假设想要创办一家公司吧！我们需要一个既酷炫又朗朗上口的名字，以及一个能赚钱的商业模式！
示例 — 公司名称及采用传统渠道的商业模式
from langchain.chains import LLMChain
from langchain.prompts import PromptTemplate
from langchain_community.llms import OpenAI
USER_INPUT = "彩色袜子"
llm = OpenAI(temperature=0)
prompt_template_product = "一家生产 的公司，起个好名字该怎么起？"
company_name_chain = LLMChain(llm=llm, prompt=PromptTemplate.from_template(prompt_template_product))
company_name_output = company_name_chain(USER_INPUT)
prompt_template_business = "请为我的公司 '" 提供最佳商业模式创意："
business_model_chain = LLMChain(llm=llm, prompt=PromptTemplate.from_template(prompt_template_business))
business_model_output = business_model_chain(company_name_output["text"])
print(company_name_output)
print(business_model_output)
>>>
>>>
这段代码很容易理解，虽然能看出一些冗余，但尚在可控范围内。
让我们通过处理用户未按预期使用该流程的情况，来增加一些自定义功能。
也许用户输入的内容与该流程的目标完全无关？在这种情况下，我们需要检测到这种情况并做出适当的响应。
示例 — 使用旧链进行自定义与路由
from langchain.chains import LLMChain
from langchain.prompts import PromptTemplate
from langchain_community.llms import OpenAI
import ast
USER_INPUT = "Harrison Chase"
llm = OpenAI(temperature=0)
# —- 与之前相同的代码
prompt_template_product = "一家生产 的公司，起个好名字该叫什么？"
company_name_chain = LLMChain(llm=llm, prompt=PromptTemplate.from_template(prompt_template_product))
prompt_template_business = "请为我的公司（名称为："
business_model_chain = LLMChain(llm=llm, prompt=PromptTemplate.from_template(prompt_template_business))
# —- 新代码
prompt_template_is_product = (
"你的目标是判断用户的输入是否为一个合理的商品名称"
"问题、问候语、长句、名人或其他无关的输入均不被视为商品名称"
"输入：n"
"回答仅限'True'或'False'，不得包含其他内容"
)
prompt_template_cannot_respond = (
"您无法对用户输入进行响应：n"
"请让用户输入一个产品名称，以便您据此创建一家公司。n"
)
cannot_respond_chain = LLMChain(llm=llm, prompt=PromptTemplate.from_template(prompt_template_cannot_respond))
company_name_chain = LLMChain(llm=llm, prompt=PromptTemplate.from_template(prompt_template_product))
business_model_chain = LLMChain(llm=llm, prompt=PromptTemplate.from_template(prompt_template_business))
is_a_product_chain = LLMChain(llm=llm, prompt=PromptTemplate.from_template(prompt_template_is_product))
# 如果对非空字符串使用 bool 运算，结果将为 True，因此我们需要使用 `literal_eval`
is_a_product = ast.literal_eval(is_a_product_chain(USER_INPUT)["text"])
if is_a_product:
company_name_output = company_name_chain(USER_INPUT)
business_model_output = business_model_chain(company_name_output["text"])
print(business_model_output)
else:
print(cannot_respond_chain(USER_INPUT))
这部分内容可能稍显晦涩，让我们来总结一下：
随之开始出现多种问题：
什么是 LangChain 表达式语言（LCEL）？
LCEL 是一种用于编写可组合且已准备就绪的链的统一接口和语法，要理解其含义，还有很多内容需要深入探讨。
我们将首先通过重写前面的链式表达式来尝试理解新的语法。
示例 — 公司名称及采用LCEL的商业模式
from langchain_core.runnables import RunnablePassthrough
from langchain.prompts import PromptTemplate
from langchain_community.llms import OpenAI
USER_INPUT = "彩色袜子"
llm = OpenAI(temperature=0)
prompt_template_product = "一家生产……的公司，起个好名字该叫什么？"
prompt_template_business = "请为我的公司（名称为：）提供一个最佳商业模式创意。"
chain = (
PromptTemplate.from_template(prompt_template_product)
| llm
|
| PromptTemplate.from_template(prompt_template_business)
| llm
)
business_model_output = chain.invoke()
短短几行代码中，却包含大量不寻常的代码：
但这种方式构建链真的更具可组合性吗？
让我们通过添加 is_a_product_chain() 函数以及在用户输入不正确时的分支逻辑来验证一下。我们甚至可以使用 Python Typing 为该链定义类型，作为一种良好的编程实践，让我们这样做吧。
示例 — 使用 LCEL 进行定制与路由
from typing import Dict
from langchain_core.runnables import RunnablePassthrough, RunnableBranch
from langchain.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain.output_parsers import BooleanOutputParser
from langchain_community.llms import OpenAI
USER_INPUT = "Harrrison Chase"
llm = OpenAI(temperature=0)
prompt_template_product = "一家生产 的公司，起个好名字该怎么起？"
prompt_template_cannot_respond = (
"您无法对用户输入进行响应：n"
"请让用户输入产品名称，以便您以此创建公司。n"
)
prompt_template_business = "请为我名为：的公司提供最佳商业模式创意"
prompt_template_is_product = (
"你的目标是判断用户输入是否为一个合理的产品名称"
"问题、问候语、长句、名人或其他无关输入均不被视为产品名称"
"输入：n"
"回答仅限'True'或'False'，不得包含其他内容"
)
answer_user_chain = (
PromptTemplate.from_template(prompt_template_product)
| llm
|
| PromptTemplate.from_template(prompt_template_business)
| llm
).with_types(input_type=Dict[str, str], output_type=str)
is_product_chain = (
PromptTemplate.from_template(prompt_template_is_product)
| llm
| BooleanOutputParser(true_val='True', false_val='False')
).with_types(input_type=Dict[str, str], output_type=bool)
cannot_respond_chain = (
PromptTemplate.from_template(prompt_template_cannot_respond) | llm
).with_types(input_type=Dict[str, str], output_type=str)
full_chain = RunnableBranch(
(is_product_chain, answer_user_chain),
cannot_respond_chain
).with_types(input_type=Dict[str, str], output_type=str)
print(full_chain.invoke())
让我们列出这些区别：
为什么LCEL更适合工业化？
如果我读到这篇文章的这一刻，有人问我是否被 LCEL 说服了，我大概会回答“不”。它的语法差异太大，而且我完全可以通过将代码组织成函数来实现几乎完全相同的代码。但我现在正在写这篇文章，所以其中一定还有其他原因。
开箱即用的调用、流式处理和批处理
使用 LCEL 后，您的区块链将自动具备以下功能：
my_chain = prompt | llm
# ———invoke——— #
result_with_invoke = my_chain.invoke("hello world!")
# ———批量——— #
result_with_batch = my_chain.batch([“hello”, “world”, “!”])
# ———流——— #
for chunk in my_chain.stream("hello world!"):
print(chunk, flush=True, end="")
在进行迭代时，您可以使用 invoke 方法来简化开发流程。但在 UI 中显示链的输出结果时，您需要以流的形式输出响应。现在，您无需重写任何代码即可使用 stream 方法。
开箱即用的异步方法
大多数情况下，应用程序的前端和后端是分离的，这意味着前端会向后端发起请求。如果有多个用户，后端可能需要同时处理多个请求。
由于 LangChain 中的大部分代码都在 API 调用之间处于等待状态，我们可以利用异步代码来提升 API 的可扩展性。如果您想了解这为何重要，建议阅读FastAPI 文档中的“并发汉堡”案例。
您无需担心具体实现，因为只要使用 LCEL，异步方法已然可用：
.ainvoke() / .abatch() / .astream：invoke、batch 和 stream 的异步版本。
我还建议阅读LangChain 文档中的“为何使用 LCEL”页面，该页面为每个同步/异步方法提供了示例。
Langchain 通过创建一个名为“Runnable”的统一接口，实现了这些“开箱即用”的功能。现在，为了充分利用 LCEL，我们需要深入了解这个新的 Runnable 接口究竟是什么。
Runnable 接口
到目前为止，我们在 LCEL 语法中使用的每个对象都是 Runnable。这是一个由 LangChain 创建的 Python 对象，该对象会自动继承我们之前讨论过的所有特性以及更多功能。通过使用 LCEL 语法，我们在每个步骤中都会组合一个新的 Runnable，这意味着最终创建的对象也将是一个 Runnable。您可以在官方文档中了解更多关于该接口的信息。
以下代码中的所有对象要么是 Runnable，要么是会被自动转换为 Runnable 的字典：
from langchain_core.runnables import RunnablePassthrough, RunnableParallel
from langchain.prompts import PromptTemplate
from langchain_community.llms import OpenAI
chain_number_one = (
PromptTemplate.from_template(prompt_template_product)
| llm
| # <— THIS WILL CHANGE
| PromptTemplate.from_template(prompt_template_business)
| llm
)
chain_number_two = (
PromptTemplate.from_template(prompt_template_product)
| llm
| RunnableParallel(company=RunnablePassthrough()) # <— THIS CHANGED
| PromptTemplate.from_template(prompt_template_business)
| llm
)
print(chain_number_one == chain_number_two)
>>> True
为什么我们要使用 RunnableParallel() 而不是直接使用 Runnable() 呢？
因为，RunnableParallel 中的每个 Runnable 都会并行执行。这意味着，如果你的 Runnable 中包含 3 个独立的步骤，它们将在机器的不同线程上同时运行，从而免费提升你的处理链的速度！
LCEL的缺点
尽管LCEL具有诸多优势，但它确实存在一些潜在的缺点：
结论
总而言之，LangChain 表达式语言（LCEL）是一款功能强大的工具，为构建 Python 应用程序带来了全新的视角。尽管其语法非传统，但我仍强烈推荐使用 LCEL，理由如下：
了解更多……
可运行的抽象
在某些情况下，我认为理解 LangChain 为实现 LCEL 语法所实现的抽象机制非常重要。
你可以按照以下方式轻松重写 Runnable 的基本功能：
class Runnable:
def __init__(self, func):
self.func = func
def __or__(self, other):
def chained_func(*args, **kwargs):
# self.func 位于左侧，other 位于右侧
return other(self.func(*args, **kwargs))
return Runnable(chained_func)
def __call__(self, *args, **kwargs):
return self.func(*args, **kwargs)
def add_ten(x):
return x + 10
def divide_by_two(x):
return x / 2
runnable_add_ten = Runnable(add_ten)
runnable_divide_by_two = Runnable(divide_by_two)
chain = runnable_add_ten | runnable_divide_by_two
result = chain(8) # 答案应为 (8+10) / 2 = 9.0
print(result)
>>> 9.0
Runnable 仅仅是一个 Python 对象，其 .__or__() 方法已被重写。
实际上，LangChain 还添加了许多功能，例如将字典转换为 Runnable、类型定义支持、可配置性支持，以及 invoke、batch、stream 和 async 方法！
那么，在您的下一个项目中，何不试用一下 LCEL 呢？
如果您想了解更多信息，我强烈建议您浏览LCEL 上的 LangChain 教程。
