管道 | 运算符常用于链接 LangChain 组件。但要构建生产级应用,需要对支持这种组合的 LangChain 表达式语言 (LCEL) 有透彻的理解。LCEL 不仅仅是语法糖;它是一种声明式地定义计算图的方式,为组件提供标准化接口,并使得流式传输、异步操作和并行执行等功能开箱即用。分析 LCEL 背后的主要思想,将帮助您充分发挥其在复杂架构中的潜力。Runnable 协议:一个标准接口LCEL 的核心是 Runnable 协议。任何旨在成为 LCEL 链一部分的组件,无论是提示模板、LLM、输出解析器还是自定义函数,都遵循这个标准接口。正是这种统一性使得不同组件能够相互连接。Runnable 协议定义了几种方法,最重要的有:invoke(input): 同步执行组件,使用给定输入。ainvoke(input): 异步执行组件。对于生产环境中的 Web 服务器上受 I/O 限制的操作(如 LLM 调用)非常重要。stream(input): 同步执行组件,并随着输出块的可用而流式返回。这对于向用户提供 LLM 的增量响应特别有用。astream(input): 异步执行组件,流式返回输出块。结合了异步执行和流式传输的优点。batch(inputs): 对输入列表同步执行组件,可能带优化。abatch(inputs): 对输入列表异步执行组件。通过对这些方法进行标准化,LCEL 确保任何 Runnable 都能以一致的方式被调用、流式处理或批量处理,无论其内部实现如何。当您实现自定义组件时(本章后面会讨论),遵循 Runnable 协议使它们在 LangChain 生态系统中成为一等公民。组合:作为 RunnableSequence 的链当您使用管道运算符 (|) 连接两个 Runnable 组件时,例如 prompt | model,LCEL 会在内部构建一个 RunnableSequence。from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI # 假设 OPENAI_API_KEY 已设置 prompt = ChatPromptTemplate.from_template("Tell me a joke about {topic}") model = ChatOpenAI() # 这会创建一个 RunnableSequence 实例 chain = prompt | model一个 RunnableSequence 本身就是一个 Runnable。当您 invoke 这个序列时,它会按顺序执行其组成组件:输入 ({"topic": "bears"}) 被传递给第一个元素 (prompt) 的 invoke 方法。第一个元素的输出(一个格式化的 PromptValue)被传递给第二个元素 (model) 的 invoke 方法。第二个元素的输出(一个 AIMessage)成为该序列的最终输出。这种顺序执行方式同样适用于 stream、astream、batch 和 abatch。对于流式传输,LCEL 处理将数据块通过序列传递。如果一个组件不原生支持流式传输(例如一个简单的提示模板),LCEL 会在其完整的输出可用后,智能地将其传递给下游,从而使得后续支持流式传输的组件(如 LLM)能够按预期运行。使用 RunnableParallel 进行并行执行LCEL 还支持 Runnable 的并行执行。这通常通过字典语法实现,该语法会创建一个 RunnableParallel 实例。from langchain_core.runnables import RunnableParallel # 示例设置(请替换为实际的 runnable) runnable1 = ChatPromptTemplate.from_template("Runnable 1: {input}") | model runnable2 = ChatPromptTemplate.from_template("Runnable 2: {input}") | model # 使用字典定义并行执行 parallel_chain = RunnableParallel(steps={ "output1": runnable1, "output2": runnable2, }) # 或者等效地: # parallel_chain = {"output1": runnable1, "output2": runnable2} # 调用此方法会潜在地并行运行 runnable1 和 runnable2 # 输入被传递给 *两个* runnable output = parallel_chain.invoke({"input": "parallel processing"}) # 输出将是一个字典: # {'output1': AIMessage(...), 'output2': AIMessage(...)}当一个 RunnableParallel(或链中使用的字典字面量)被调用时,它会将 相同 的输入传递给其包含的每个 Runnable。然后,它会尝试并发执行这些 Runnable,尤其是在使用 ainvoke 或 abatch 等异步方法时。最终输出是一个字典,其中键与定义中提供的键相对应,值是相应 Runnable 的输出。digraph G { rankdir=LR; node [shape=box, style=rounded, fontname="sans-serif", fillcolor="#e9ecef", style=filled]; edge [fontname="sans-serif"]; subgraph cluster_seq { label = "顺序执行 (RunnableSequence)"; bgcolor="#f8f9fa"; style=filled; s_input [label="输入"]; s_r1 [label="Runnable 1"]; s_r2 [label="Runnable 2"]; s_output [label="输出"]; s_input -> s_r1; s_r1 -> s_r2 [label="输出 1"]; s_r2 -> s_output [label="输出 2"]; } subgraph cluster_par { label = "并行执行 (RunnableParallel)"; bgcolor="#f8f9fa"; style=filled; p_input [label="输入"]; p_r1 [label="Runnable A", fillcolor="#a5d8ff"]; p_r2 [label="Runnable B", fillcolor="#a5d8ff"]; p_output [label="输出\n{A_输出, B_输出}"]; p_input -> p_r1; p_input -> p_r2; p_r1 -> p_output [label="输出 A"]; p_r2 -> p_output [label="输出 B"]; } }LCEL 中顺序 (RunnableSequence) 和并行 (RunnableParallel) 执行模式下的数据流比较。理解这种区别对于设计高效的链具有重要意义。当一个步骤的输出是下一个步骤的输入时,使用 RunnableSequence (|)。当多个步骤可以独立地且可能并发地处理相同输入时,使用 RunnableParallel ({})。配置和流式传输的内部机制LCEL 在任何 Runnable 上都提供了一个 .with_config() 方法,用于传递配置选项,例如回调、递归限制或运行名称。此配置通常会通过序列或并行执行图进行传播。# 传递配置的示例 result = chain.with_config( {"run_name": "JokeGenerationRun"} ).invoke({"topic": "robots"})流式传输支持是一个核心功能。当您在 LCEL 链上调用 .stream() 或 .astream() 时,框架会协调端到端的流式传输过程。如果 LLM 组件流式传输令牌,LCEL 会确保这些令牌在到达时被生成。对于序列,中间的非流式传输步骤会完成执行,其输出被传递给下一个步骤,然后该步骤可能会启动流式传输。为流式传输设计的输出解析器可以增量处理令牌块。这种内置的流式传输能力对于用户期望即时反馈的交互式应用很重要。对高级用法的意义理解 LCEL 内部机制不仅仅是学术上的事。它直接影响您进行以下操作的能力:高效调试: 理解 Runnable 接口以及 RunnableSequence 和 RunnableParallel 如何执行,有助于追踪数据流并找出错误或意外行为,尤其是在使用 LangSmith 等工具时(第 5 章中会讲到)。优化性能: 识别并行执行 (RunnableParallel) 的机会或运用异步方法 (ainvoke、astream、abatch),使您能够构建更具响应性和可扩展性的应用程序(在第 6 章中会进一步探讨)。实现自定义组件: 当您需要内置 LangChain 组件之外的功能时,实现 Runnable 协议可确保您的自定义逻辑集成到 LCEL 链中,继承流式传输和配置管理等功能。构建复杂逻辑: 将顺序和并行执行与配置结合起来,可以在声明式且易于管理的框架内创建复杂的、多步骤的推理过程或代理行为。LCEL 为复杂的 LangChain 应用提供了架构支柱。通过理解其核心原理、Runnable 协议、通过 RunnableSequence 进行的顺序组合、通过 RunnableParallel 进行的并行执行、配置传播以及对流式传输和异步操作的固有支持,您将更有能力设计、构建和调试高级的、生产级的 LLM 系统。