LCEL 链式表达语言：声明式构建 LangChain 流水线

FreeGuideOnline 最新 2026-06-14

LCEL 链式表达语言：声明式构建 LangChain 流水线

概述

LCEL（LangChain Expression Language）是 LangChain 中用于声明式构建链式流水线的核心表达语言。它允许你通过简洁的管道操作符（|）和丰富的内置原语，将提示模板、语言模型、输出解析器等组件快速组合成复杂的处理链条，而无需编写大量胶水代码。

与传统的命令式链相比，LCEL 具有以下优势：

  • 可读性极强：代码结构清晰，一眼就能看出数据流向。
  • 自动并行：支持在联合节点（如 RunnableParallel）中自动并行调用。
  • 流式支持：所有用 LCEL 定义的链自然支持流式传输（token by token）。
  • 可追踪与调试：每条链都会自动记录调用历史，便于监控和排查。
  • 可组合性：链可以像搭积木一样任意嵌套复用。

环境准备

在使用 LCEL 前，请确保已安装 langchain 及相关依赖：

pip install langchain langchain-core langchain-openai

本教程中的示例将使用 OpenAI 模型，你需要设置 API 密钥：

import os
os.environ["OPENAI_API_KEY"] = "your-api-key"

基础语法

Runnable 协议

LCEL 中的一切操作都围绕 Runnable 接口展开。任何能够接受输入并产生输出的对象都是 Runnable。LangChain 中的提示模板、模型、解析器、函数等都已实现了该接口。

最核心的 Runnable 方法是 invoke（单次调用）、stream（流式调用）和 batch（批量调用）。

管道操作符 |

LCEL 使用 Unix 管道符号 | 将两个 Runnable 首尾相连。a | b 会创建一个新的 Runnable，其执行流程为：

  1. 调用 Runnable a，得到输出 output_a。
  2. 将 output_a 作为输入传递给 Runnable b。
  3. 返回 b 的结果。

示例：简单的提示 + 模型链

from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI

prompt = ChatPromptTemplate.from_template("讲一个关于{topic}的笑话")
model = ChatOpenAI(model="gpt-3.5-turbo")

chain = prompt | model   # LCEL 链

result = chain.invoke({"topic": "程序员"})
print(result.content)

这里 prompt 和 model 都是 Runnable，管道将它们串联，invoke 时输入字典会自动传递给 prompt，其输出字符串则作为 model 的输入。

绑定参数：bind 与 with_config

一些组件需要附加固定参数，例如给模型设置 temperature。

model_with_params = ChatOpenAI(model="gpt-4").bind(temperature=0.7)
chain = prompt | model_with_params

bind 返回一个新的 Runnable，它会在调用时自动携带预设参数。

数据变换与增强

自定义函数：RunnableLambda

你可以将普通 Python 函数包装成 Runnable，无缝接入管道。

from langchain_core.runnables import RunnableLambda

def add_hello(text: str) -> str:
    return "你好！" + text

runnable_func = RunnableLambda(add_hello)
chain = prompt | model | runnable_func

RunnableLambda 会自动根据函数的签名处理输入输出。对于更复杂的转换，也可使用 RunnablePassthrough。

透传与选择器：RunnablePassthrough 和 RunnablePick

有时你希望在链中保留上游的输出，同时附加新数据。

from langchain_core.runnables import RunnablePassthrough, RunnablePick

# 保留整个输入，并添加额外键
chain_with_extra = RunnablePassthrough.assign(
    joke=lambda x: x["text"]   # 假设上游输出包含 'text'
)

RunnablePick 用于从输入字典中选择特定键：

pick_key = RunnablePick("question")
# 会将 {"question": "你好"} → "你好"

结构化输出：output_parser

常见的设计是将模型输出直接解析为结构化对象。

from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate

prompt = ChatPromptTemplate.from_template("列出{count}种宠物，用逗号分隔。")
model = ChatOpenAI()
parser = StrOutputParser()

chain = prompt | model | parser
result = chain.invoke({"count": 5})
print(result)  # 直接得到字符串

高级组合模式

并行执行：RunnableParallel（字典并联）

当需要同时调用多个子链时，使用字典定义并行结构。

from langchain_core.runnables import RunnableParallel

joke_chain = (
    ChatPromptTemplate.from_template("一个关于{topic}的笑话")
    | ChatOpenAI()
    | StrOutputParser()
)
poem_chain = (
    ChatPromptTemplate.from_template("一首关于{topic}的诗")
    | ChatOpenAI()
    | StrOutputParser()
)

map_chain = RunnableParallel(joke=joke_chain, poem=poem_chain)
result = map_chain.invoke({"topic": "月亮"})
print(result["joke"])
print(result["poem"])

字典中的每个值都是一个独立的Runnable，它们会自动并行执行（如果底层执行器支持），最终结果合并为一个字典。

你也可以在现有链上使用 assign 方法动态附加并行分支：

base_chain = prompt | model | StrOutputParser()
chain_with_translation = base_chain.assign(
    translation=lambda input_dict: translate_chain.invoke(input_dict["output"])
)

条件分支：RunnableBranch

当需要根据输入内容动态选择不同的链时，使用 RunnableBranch。

from langchain_core.runnables import RunnableBranch

branch = RunnableBranch(
    (lambda x: "紧急" in x["text"], urgent_chain),
    (lambda x: "正常" in x["text"], normal_chain),
    default_chain   # 不满足任何条件时执行
)

full_chain = prompt | model | StrOutputParser() | branch

每个条件是一个 (条件函数, Runnable) 元组。第一个条件成立则执行对应分支，否则继续向下匹配。

链式组合：chain1.with_fallbacks(chain2)

容错机制，当主链调用失败时自动切换到备选链。

primary_chain = ChatOpenAI(model="gpt-4") | parser
backup_chain = ChatOpenAI(model="gpt-3.5-turbo") | parser

robust_chain = primary_chain.with_fallbacks([backup_chain])

动态路由：RunnableRouter（高级）

对于更动态的分发场景，可以自定义路由逻辑，但这通常已包含在 RunnableBranch 中。如需极致灵活，可自行实现函数返回 Runnable。

流式与批量调用

所有 LCEL 链原生支持流式输出：

for chunk in chain.stream({"topic": "人工智能"}):
    print(chunk, end="", flush=True)  # token 级别的实时打印

批量调用自动优化并发：

inputs = [{"topic": "程序员"}, {"topic": "医生"}, {"topic": "律师"}]
results = chain.batch(inputs)   # 内部自动并行
# 可配置并发数：chain.batch(inputs, config={"max_concurrency": 3})

追溯与调试

LangChain 提供了丰富的回调机制。通过设置全局追踪，你可以在 LangSmith 等平台查看每一步的输入输出和执行耗时。同时，chain.get_graph().print_ascii() 可以打印链的 ASCII 流程图：

+-------------+
| PromptInput |
+-------------+
       |
       v
+-------------+
| ChatPrompt  |
+-------------+
       |
       v
+-------------+
| ChatOpenAI  |
+-------------+
       |
       v
+-------------+
| StrOutputPar|
+-------------+

完整示例：客服意图识别流水线

以下是一个综合应用，根据用户消息判断意图（投诉、咨询、闲聊），并分别给出回复。

from langchain_core.runnables import RunnableParallel, RunnableBranch, RunnablePassthrough
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI

model = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)

# 意图分类器
classify_prompt = ChatPromptTemplate.from_template(
    "判断以下客户消息的意图，仅回复一个词：投诉、咨询、闲聊。\n消息：{input}"
)
classify_chain = classify_prompt | model | StrOutputParser()

# 三种不同回应链
complaint_chain = (
    ChatPromptTemplate.from_template("你是一个专业客服，针对用户投诉给出安抚和解决方案。投诉内容：{input}")
    | model | StrOutputParser()
)
inquiry_chain = (
    ChatPromptTemplate.from_template("你是一个知识库，回答用户的问题。问题：{input}")
    | model | StrOutputParser()
)
chitchat_chain = (
    ChatPromptTemplate.from_template("用友好幽默的语气回复用户的闲聊。内容：{input}")
    | model | StrOutputParser()
)

# 根据意图分支
intent_branch = RunnableBranch(
    (lambda x: "投诉" in x["intent"], complaint_chain),
    (lambda x: "咨询" in x["intent"], inquiry_chain),
    chitchat_chain   # 默认闲聊
)

# 最终链：先分类，并将结果和原始输入一同传递给分支
final_chain = RunnableParallel(
    intent=classify_chain,
    input=RunnablePassthrough()   # 传递原始输入
) | intent_branch

# 测试
print(final_chain.invoke("我刚买的手机屏幕就坏了，你们怎么处理？"))  # 触发投诉链

自定义 LCEL 组件（进阶）

你可以通过继承 Runnable 创建自定义组件。只需实现 invoke 方法（必要时实现 stream 和 batch 以获取性能优化）。几乎所有 LangChain 组件都遵循这一模式，因此自定义单位可以完美融入现有生态。

from langchain_core.runnables import Runnable

class ToUpperCase(Runnable):
    def invoke(self, input: str, config=None) -> str:
        return input.upper()

custom = ToUpperCase()
chain = prompt | model | StrOutputParser() | custom

总结

  • LCEL 通过 | 管道和 Runnable 协议将构建复杂流水线的成本降到最低。
  • RunnableParallel 实现无痛并行，RunnableBranch 处理条件逻辑，RunnablePassthrough 传递数据。
  • 所有链天然支持流式、批量、回溯，适合生产环境。
  • 只需掌握几个原语，你就能从简单原型快速演进到复杂的企业级 LLM 应用。

立即动手，将你原有的 LangChain 代码改写成 LCEL 风格，感受声明式链式表达带来的清晰与高效！