6. 使用 LCEL 构建声明式管道

在前面的章节中,我们一直在编写命令式代码来编排 LLM 交互:创建提示词、调用模型、解析响应。这种方式可行,但随着 AI 应用变得越来越复杂,这种方法会变得冗长且难以维护。你最终会得到深度嵌套的函数调用、每一步的手动错误处理,以及难以理解的整体数据流。

LangChain 表达式语言(LCEL)通过让你声明想要发生什么,而不是如何实现它来解决这个问题。你不需要编写按顺序调用函数的过程式代码,而是使用简单的管道操作符(|)来组合组件,读起来就像 Unix 管道一样。结果是更清晰、更易维护的代码,能够清楚地表达数据在 AI 系统中的流动。

本章介绍如何使用 LCEL 构建线性工作流 - 数据从开始到结束流动的操作序列,没有分支或循环。我们将介绍何时使用 LCEL、如何组合管道,以及如何同步执行和流式输出。

6.1) 为什么使用 LCEL?

LCEL 解决的问题

让我们从一个具体的例子开始。假设你正在构建一个客户支持助手,需要:

1. 接收用户问题
2. 将其格式化为包含系统指令和用户消息的提示词
3. 发送给 LLM
4. 解析响应以提取文本内容

以下是你可能如何命令式地编写这个流程(不使用 LCEL):

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
 
llm = ChatOpenAI(model="gpt-4o-mini")
 
prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful customer support assistant."),
    ("user", "{question}")
])
 
parser = StrOutputParser()
 
def answer_question(question: str) -> str:
    # 步骤 1: 使用问题格式化提示词
    messages = prompt.invoke({"question": question})
    
    # 步骤 2: 调用 LLM
    response = llm.invoke(messages)
    
    # 步骤 3: 解析输出以提取文本内容
    result = parser.invoke(response)
    
    return result
 
# 使用它
result = answer_question("How do I reset my password?")
print(result)

输出:

要重置您的密码,请按照以下步骤操作:
1. 前往登录页面
2. 点击"忘记密码"
3. 输入您的电子邮件地址
4. 检查您的电子邮件以获取重置链接
5. 点击链接并创建新密码

这种方式可行,但请注意以下问题:

冗长: 每一步都需要显式的变量赋值和函数调用。实际的逻辑(格式化 → 调用 → 解析)被埋在样板代码中。

结构僵化: 如果你想添加一个步骤(比如验证问题或记录响应),你需要在函数中间插入代码,增加复杂性。

没有内置流式输出: 要流式传输令牌,你需要重写整个函数以使用 llm.stream() 并手动处理异步迭代。

数据流不清晰: 阅读代码时,不能立即看出这是一个简单的管道。你必须追踪变量赋值才能理解流程。

现在让我们看看 LCEL 版本:

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
 
# 定义组件(与之前相同)
llm = ChatOpenAI(model="gpt-4o-mini")
 
prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful customer support assistant."),
    ("user", "{question}")
])
 
parser = StrOutputParser()
 
# 使用管道操作符将它们组合成链
chain = prompt | llm | parser
 
# 使用它
result = chain.invoke({"question": "How do I reset my password?"})
print(result)

输出:

要重置您的密码,请按照以下步骤操作:
1. 前往登录页面
2. 点击"忘记密码"
3. 输入您的电子邮件地址
4. 检查您的电子邮件以获取重置链接
5. 点击链接并创建新密码

输出是相同的,但代码却截然不同:

声明式: chain = prompt | llm | parser 在一行中表达了整个流程。从左到右阅读:提示词 → LLM → 解析器。

可组合: 每个组件(prompt、llm、parser)都是独立且可重用的。你可以在不重写管道的情况下交换组件。

内置流式输出: 在 chain.invoke() 和 chain.stream() 之间切换,无需更改管道定义。

意图清晰: | 操作符使数据流一目了然。

LCEL 的适用场景:线性工作流

LCEL 专为线性工作流设计 - 数据从开始到结束单向流动的序列,没有分支或循环。

这种线性模式涵盖了许多 AI 应用。考虑一个文档问答系统:你接收一个问题 → 检索相关文档 → 格式化提示词 → 发送给 LLM → 解析答案。每一步都是清晰的序列,其中一步的输出成为下一步的输入。

但如果你的工作流需要:

• 让 LLM 根据问题决定调用哪个工具
• 调用工具,观察结果,然后决定下一步做什么
• 重试失败的操作,使用不同的方法

这些场景需要循环和条件分支 - LCEL 无法处理的事情。这就是为什么存在 LangGraph(我们将在第 15 章介绍它)。

以下是两种方法的可视化比较:

LCEL 非常适合直接的管道,其中每一步处理前一步的输出。数据只向一个方向流动。

LangGraph 用于需要决策和循环的场景。代理可以行动、观察结果,然后再次思考。

在本章中,我们专注于 LCEL。为什么要先掌握线性管道?

• 基础: 管道操作符(|)是 LangChain 的核心语法。理解这一点会让其他一切变得更容易。
• LangGraph 的前提: LangGraph 代理在其节点内部广泛使用 LCEL 链
• 实际模式: 复杂的代理是通过组合 LCEL 链构建的

本章的其余部分将向你展示如何使用管道操作符(|)构建这些线性管道。

6.2) 使用 | 操作符组合管道

LCEL 管道的工作原理

LCEL 的管道操作符(|)之所以有效,是因为 Runnable 接口。

什么是 Runnable?

Runnable 是 LangChain 的标准接口。当一个组件实现 Runnable 接口时,它可以使用 | 操作符与其他组件链接。

每个 Runnable 都提供这些方法:

• .invoke(input) - 执行一次并获得完整结果
• .stream(input) - 执行并在 LLM 生成令牌时接收每个词
• .batch(inputs) - 使用不同的输入执行多次并获得所有结果

在本章中,我们将专注于 .invoke() 和 .stream()(我们将在需要时稍后介绍 .batch())。

为什么 | 有效?

因为 Runnable 类使用 Python 的操作符重载实现了 | 操作符。当你编写 prompt | llm 时,它会创建一个新的 Runnable,按顺序执行这两个组件。

大多数 LangChain 组件都是 Runnable

这就是为什么你可以链接这么多不同的组件:

• ChatPromptTemplate 是一个 Runnable
• ChatOpenAI 是一个 Runnable
• StrOutputParser 是一个 Runnable
• 甚至你用 | 创建的自定义链本身也是 Runnable!

这意味着你可以通过组合更简单的管道来构建复杂的管道。

连接组件:输入/输出类型

使用 | 连接组件时,你需要确保一个组件的输出类型与下一个组件的输入类型匹配。

关键组件签名:

示例流程:

chain = prompt | llm | parser

以下是数据类型在通过每个组件时的转换:

• prompt: 接收 dict 作为输入并转换为 list[BaseMessage]
• llm: 接收 list[BaseMessage] 作为输入并转换为 AIMessage
• parser: 接收 AIMessage 作为输入并转换为 str

让我们看看实际操作:

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
 
prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant."),
    ("user", "{question}")
])
 
llm = ChatOpenAI(model="gpt-4o-mini")
parser = StrOutputParser()
 
chain = prompt | llm | parser
 
result = chain.invoke({"question": "What is 2+2?"})
print(result)  # "2+2 等于 4。"

每一步发生了什么:

如果类型不匹配会发生什么?

如果你尝试连接不兼容的组件,你会得到一个错误:

# 错误: 这不会工作
chain = llm | prompt  # LLM 输出 AIMessage,但 prompt 需要 dict 作为输入

错误消息会告诉你下一个组件期望的输入类型与它实际接收到的输入类型。

6.3) 执行链:.invoke() 和 .stream()

运行管道

一旦你组合了一个链,你可以使用 .invoke() 方法执行它。这是运行管道的同步方式 - 它会等待整个响应后再返回结果。

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
 
prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant."),
    ("user", "{question}")
])
 
llm = ChatOpenAI(model="gpt-4o-mini")
chain = prompt | llm | StrOutputParser()
 
# 同步调用
result = chain.invoke({"question": "What is 2+2?"})
print(result)

输出:

2+2 等于 4。

.invoke() 方法很简单:传入第一个组件期望的输入参数,并获得最后一个组件产生的输出。

从同一链流式传输令牌

虽然 .invoke() 很简单,但它对面向用户的应用有一个限制:用户在整个响应完成之前什么都看不到。对于长响应(10-20 秒),这会造成糟糕的用户体验。

流式输出在令牌生成时立即显示它们,而不是等待完整的响应。这就是你在 ChatGPT 中看到的打字效果。

让我们看看流式输出的实际操作:

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
 
prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant."),
    ("user", "{question}")
])
 
llm = ChatOpenAI(model="gpt-4o-mini")
chain = prompt | llm | StrOutputParser()
 
# 在块到达时流式传输
for chunk in chain.stream({"question": "Explain what Python is in one sentence"}):
    print(chunk, end="", flush=True)

输出(实时显示,逐个令牌):

Python 是一种多功能的高级编程语言,以其简洁性和可读性而闻名,广泛用于 Web 开发、数据科学和自动化。

请注意,链定义与上面的 .invoke() 示例完全相同。我们不需要重建它 - 我们只是调用了 .stream() 而不是 .invoke()。

.stream() 方法在 LLM 生成令牌时以块的形式返回结果。print(chunk, end="", flush=True) 立即在屏幕上显示每个块,创建实时打字效果。

何时使用 .invoke() 与 .stream()

使用 .invoke() 当:

• 你只需要最终结果(分析、翻译、分类)
• 响应很短,等待时间不是问题
• 你需要完整的输出才能继续下一步

使用 .stream() 当:

• 用户需要看到进度(聊天、内容生成)
• 响应很长,等待时间会很明显
• 你正在构建一个实时反馈很重要的 UI

两种方法都适用于同一个链。定义一次链,然后根据你的需求选择执行模式。

在本章中,你学习了 LCEL - LangChain 的声明式管道语法:

• 构建管道使用管道操作符:prompt | llm | parser
• 灵活执行:使用 .invoke() 获取完整结果或使用 .stream() 获取实时输出
• 类型安全:链接组件时将输出类型与输入类型匹配

同一个链适用于两种执行模式 - 定义一次,随处使用。

下一章: 第 7 章介绍使用 Pydantic 的结构化输出,使你能够从 LLM 响应中提取经过验证的 JSON 数据。