利用 LangChain 表达式语言（LCEL）和 Runnables 构建智能、模块化 LLM 工作流

在当今人工智能领域，构建高效、可维护的 LLM 工作流至关重要。LangChain 提供的 LangChain 表达式语言（LCEL） 和 Runnables 正是实现这一目标的强大工具。本文将深入探讨 LCEL 和 Runnables 的概念，并通过实际案例展示如何使用它们构建智能、模块化的 LLM 应用。

LCEL：声明式、可组合的 LLM 管道

传统的 LLM 应用开发通常需要编写大量的模板代码来连接不同的组件。LangChain 表达式语言（LCEL） 旨在简化这一过程，它提供了一种声明式、可组合的方式来构建 LLM 管道。与传统的 LLMChain 相比，LCEL 使用管道操作符 | 将不同的组件像函数一样连接起来，例如：chain = prompt | llm | parser。这种方式不仅代码更简洁、易读，而且更易于调试和复用。

例如，假设我们需要构建一个简单的翻译流程，将英文翻译成中文。使用传统的 LLMChain 可能需要编写如下代码：

from langchain.chains import LLMChain
from langchain.prompts import PromptTemplate
from langchain.llms import OpenAI

prompt = PromptTemplate(
    input_variables=["text"],
    template="Translate '{text}' to Chinese."
)

llm = OpenAI(temperature=0.7)

chain = LLMChain(prompt=prompt, llm=llm)

result = chain.run("Hello, world!")
print(result)

而使用 LCEL，我们可以用更简洁的方式实现相同的功能：

from langchain.prompts import PromptTemplate
from langchain.llms import OpenAI
from langchain.output_parsers import StrOutputParser

prompt = PromptTemplate.from_template("Translate to Chinese: {sentence}")
llm = OpenAI(temperature=0.7)
parser = StrOutputParser()

chain = prompt | llm | parser

output = chain.invoke({"sentence": "Hello, world!"})
print(output)

可以明显看出，LCEL 的代码更加简洁易懂，也更容易维护。

Runnables：构成 LCEL 的基石

Runnables 是 LangChain 中构建 LLM 工作流 的基本构建块。它们是可组合的组件，可以像乐高积木一样拼接在一起，构建复杂的逻辑流程。LangChain 提供了多种内置的 Runnables，包括：

RunnableSequence: 按顺序连接多个 Runnables，将前一个的输出传递给下一个。
RunnableLambda: 将任何 Python 函数封装成一个 Runnable。
RunnablePassthrough: 原样返回输入，用于调试或占位符步骤。
RunnableParallel: 并行运行多个 Runnables，并返回结果字典。
RunnableBranch: 根据条件将输入路由到不同的路径。

这些 Runnables 提供了强大的灵活性，可以满足各种不同的 LLM 应用 需求。

1. RunnableSequence：线性流程的核心

RunnableSequence 是最基本的 Runnable，它将多个 Runnables 按顺序连接起来，形成一个线性流程。例如，我们可以使用 RunnableSequence 将一个 Prompt、一个 LLM 和一个 Parser 连接起来，构建一个完整的 LLMChain。

from langchain_core.runnables import RunnableSequence
from langchain.prompts import PromptTemplate
from langchain.llms import OpenAI
from langchain.output_parsers import StrOutputParser

prompt = PromptTemplate.from_template("Translate to Spanish: {sentence}")
llm = OpenAI(temperature=0.7)
parser = StrOutputParser()

sequence = RunnableSequence(first=prompt, middle=llm, last=parser)

result = sequence.invoke({"sentence": "How are you?"})
print(result)

这段代码定义了一个翻译流程，首先使用 PromptTemplate 将输入转换为提示，然后使用 OpenAI LLM 进行翻译，最后使用 StrOutputParser 将输出解析为字符串。RunnableSequence 保证了这些步骤按顺序执行，从而实现了一个完整的翻译功能。

2. RunnableLambda：自定义逻辑的利器

RunnableLambda 允许我们将任何 Python 函数封装成一个 Runnable，从而在 LLM 管道 中添加自定义逻辑。这使得我们可以灵活地处理数据、进行验证或执行其他任何需要的操作。

例如，我们可以使用 RunnableLambda 从一段文本中提取姓名：

from langchain_core.runnables import RunnableLambda

def extract_name(text: str) -> str:
    try:
        return text.split("My name is")[-1].strip().split()[0]
    except IndexError:
        return "Unknown"

name_extractor = RunnableLambda(extract_name)

print(name_extractor.invoke("Hi! My name is Karan Sharma."))
print(name_extractor.invoke("Hello! My name is.")) #test case with no name after statement

这段代码定义了一个 extract_name 函数，用于从包含 “My name is” 的文本中提取姓名。然后，我们使用 RunnableLambda 将这个函数封装成一个 Runnable，并可以使用 invoke 方法来执行它。通过捕获 IndexError 异常，我们使得这个函数更加健壮。

3. RunnablePassthrough：调试和占位符的助手

RunnablePassthrough 是一个非常简单的 Runnable，它原样返回输入。它主要用于调试和占位符步骤。在调试复杂的 LLM 管道 时，我们可以使用 RunnablePassthrough 来查看中间步骤的输出，从而更容易发现问题。

from langchain_core.runnables import RunnablePassthrough

passthrough = RunnablePassthrough()

print(passthrough.invoke("Send this through!"))

4. RunnableParallel：并行处理的加速器

RunnableParallel 允许我们并行运行多个 Runnables，并返回一个包含所有结果的字典。这可以显著提高 LLM 工作流 的效率，尤其是在需要执行多个独立任务时。

例如，我们可以使用 RunnableParallel 同时翻译一段文本并将其转换为大写：

from langchain_core.runnables import RunnableParallel, RunnableLambda
from langchain.prompts import PromptTemplate
from langchain.llms import OpenAI
from langchain.output_parsers import StrOutputParser

prompt = PromptTemplate.from_template("Translate to French: {sentence}")
llm = OpenAI(temperature=0.7)
parser = StrOutputParser()

chain1 = prompt | llm | parser
chain2 = RunnableLambda(lambda x: x['sentence'].upper())

parallel = RunnableParallel({
    "translated": chain1,
    "shouted": chain2
})

output = parallel.invoke({"sentence": "Good morning"})
print(output)

这段代码定义了两个 Runnables：chain1 用于将文本翻译成法语，chain2 用于将文本转换为大写。然后，我们使用 RunnableParallel 将这两个 Runnables 并行运行，并将结果存储在一个字典中。

5. RunnableBranch：条件逻辑的导航器

RunnableBranch 允许我们根据条件将输入路由到不同的路径。这使得我们可以根据不同的输入执行不同的逻辑，从而构建更灵活的 LLM 应用。

例如，我们可以使用 RunnableBranch 根据输入是否包含问号来执行不同的操作：

from langchain_core.runnables import RunnableBranch, RunnableLambda

is_question = lambda x: "?" in x["text"]
branch = RunnableBranch([
    (is_question, RunnableLambda(lambda x: "That's a question.")),
    (lambda x: True, RunnableLambda(lambda x: "That's a statement."))
])

print(branch.invoke({"text": "Is this AI?"}))
print(branch.invoke({"text": "LangChain is great"}))

这段代码定义了一个 is_question 函数，用于判断输入是否包含问号。然后，我们使用 RunnableBranch 将输入路由到不同的 Lambda 函数，根据输入是问题还是陈述返回不同的结果.

整合所有组件：构建完整的 LLM 应用

现在，让我们将所有这些 Runnables 组合起来，构建一个完整的 LLM 应用。我们将构建一个流程，该流程接受用户输入，将其翻译成印地语，提取姓名，并根据输入是问题还是陈述返回不同的响应。

from langchain.prompts import ChatPromptTemplate
from langchain.output_parsers import StrOutputParser
from langchain.llms import OpenAI
from langchain_core.runnables import (
    RunnableLambda, RunnableBranch, RunnablePassthrough, RunnableParallel
)

llm = OpenAI(temperature=0.7)
parser = StrOutputParser()
prompt = ChatPromptTemplate.from_template("Translate to Hindi: {sentence}")

# Step 1: Translate
translate = prompt | llm | parser

# Step 2: Extract name
extract_name = RunnableLambda(lambda x: x.split("मेरा नाम")[-1].split()[0] if "मेरा नाम" in x else "Unknown")

# Step 3: Branch output
is_question = lambda x: "?" in x["sentence"]
branch = RunnableBranch([
    (is_question, RunnableLambda(lambda x: "You're asking something.")),
    (lambda x: True, RunnableLambda(lambda x: "You're saying something."))
])

# Final parallel chain
pipeline = RunnableParallel({
    "translated": translate,
    "name": translate | extract_name,
    "type": branch
})

result = pipeline.invoke({"sentence": "मेरा नाम रोहित है।"})
print(result)

这个流程首先将用户输入翻译成印地语，然后尝试提取姓名，最后根据输入是问题还是陈述返回不同的响应。整个流程使用 LCEL 和 Runnables 构建，代码简洁易懂，易于维护和扩展。

更多 LCEL 示例

以下是一些使用 LangChain 表达式语言（LCEL） 的更多示例，以帮助你更好地理解其用法：

基本示例：LLMChain 与 LCEL

from langchain.prompts import ChatPromptTemplate
from langchain.llms import OpenAI
from langchain.output_parsers import StrOutputParser

llm = OpenAI(temperature=0.7)
prompt = ChatPromptTemplate.from_template("Translate to Hindi: {sentence}")
parser = StrOutputParser()

# 使用 LCEL 定义 chain
chain = prompt | llm | parser

# 运行 chain
output = chain.invoke({"sentence": "Good morning"})
print(output)

Chain 组合：Prompt → Gemini → JSON Parser

from langchain.output_parsers import StructuredOutputParser, ResponseSchema
from langchain.prompts import ChatPromptTemplate
from langchain.llms import OpenAI

schemas = [
    ResponseSchema(name="name", description="Full name"),
    ResponseSchema(name="occupation", description="Job or role")
]
parser = StructuredOutputParser.from_response_schemas(schemas)
prompt = ChatPromptTemplate.from_template(
    "Extract structured info:\n\n{text}\n\n{format_instructions}")
llm = OpenAI(temperature=0.7)
chain = prompt | llm | parser
result = chain.invoke({
    "text": "My name is Neha, and I work as a UX designer.",
    "format_instructions": parser.get_format_instructions()
})
print(result)

使用 RunnableSequence 的多步骤工作流

from langchain_core.runnables import RunnableLambda
from langchain.output_parsers import StrOutputParser
from langchain.prompts import ChatPromptTemplate
from langchain.llms import OpenAI

def extract_name(text: str) -> str:
    # 从输入中提取姓名的简单函数
    if "My name is" in text:
        return text.split("My name is")[-1].split(".")[0].strip()
    return "Unknown"

llm = OpenAI(temperature=0.7)
translate_prompt = ChatPromptTemplate.from_template("Translate to Hindi: {sentence}")
translate_chain = translate_prompt | llm | StrOutputParser()
name_chain = RunnableLambda(extract_name)
final_chain = translate_chain | name_chain
print(final_chain.invoke({"sentence": "My name is Raghav."}))

重用 Chains (RunnableConfig & Batch Support)

from langchain_core.runnables import RunnableConfig

# 批量运行
# chain.batch([
#     {"sentence": "How are you?"},
#     {"sentence": "What is your name?"}
# ])

# 使用配置进行跟踪/调试
config = RunnableConfig(tags=["translate"], metadata={"lang": "Hindi"})
# Assuming 'chain' is defined as in previous examples
# chain.invoke({"sentence": "Good night"}, config=config)

总结

LangChain 表达式语言（LCEL） 和 Runnables 为构建智能、模块化的 LLM 工作流 提供了强大的工具。LCEL 提供了一种声明式、可组合的方式来定义 LLM 管道，而 Runnables 则提供了构建管道的基本构建块。通过组合不同的 Runnables，我们可以构建各种复杂的 LLM 应用，例如翻译、信息提取、问答等等。

掌握 LCEL 和 Runnables 对于任何希望构建高效、可维护的 LLM 应用 的开发者来说都是至关重要的。通过本文的学习，相信你已经对 LCEL 和 Runnables 有了深入的了解，可以开始使用它们构建自己的 LLM 工作流 了。在构建 LLM 应用 时，务必考虑使用 RunnableBranch， RunnableLambda 和 RunnableParallel 表达式来获得更强大的逻辑表达能力。

利用 LangChain 表达式语言（LCEL）和 Runnables 构建智能、模块化 LLM 工作流