LangChain表达式语言（LCEL）与Runnable协议深度解析

引言

如果你在过去两年里构建过基于大语言模型（LLM）的应用，大概率遇到过这样的场景：你花了一个周末精心设计了一个包含提示词模板、模型调用、输出解析的完整流程，代码逻辑清晰、运行完美。然而当需求从“单轮问答”演变为“多步推理”、“条件分支”或“流式输出”时，你发现自己陷入了一场重构的噩梦——原本简洁的prompt -> llm -> parser三段式代码，被硬生生改造成了复杂的回调函数地狱和状态同步逻辑。

这正是LangChain表达式语言（LCEL）试图终结的痛点。作为LangChain生态在2024年之后的核心抽象，LCEL引入的Runnable协议彻底改变了链路的构建方式。本文将带领你深入LCEL的源码肌理，理解其设计哲学，并给出真正能落地的实战方案。

核心概念：从“管道”到“协议”

生活类比：乐高积木与标准接口

想象你在搭建一座乐高城堡。传统的LangChain调用方式，是使用一块块已经粘死的“复合积木”——每个积木内部有自己的逻辑，但接口千奇百怪。你需要用胶水（自定义胶水代码）将它们强行连接。而LCEL则定义了乐高积木的标准凸起与凹槽——这就是Runnable协议。任何实现了这个协议的对象，无论内部是提示模板、模型还是自定义函数，都能以完全相同的方式被组合、并行、回退和流式输出。

技术定义

Runnable协议的核心是一组异步和同步方法的规范：

class Runnable(Generic[Input, Output]):
    def invoke(self, input: Input, config: Optional[RunnableConfig] = None) -> Output: ...
    async def ainvoke(self, input: Input, config: Optional[RunnableConfig] = None) -> Output: ...
    def batch(self, inputs: List[Input], config: Optional[RunnableConfig] = None) -> List[Output]: ...
    def stream(self, input: Input, config: Optional[RunnableConfig] = None) -> Iterator[Output]: ...
    def bind(self, **kwargs) -> Runnable[Input, Output]: ...
    # 组合运算符：| 是核心
    def __or__(self, other: Runnable) -> Runnable: ...

LCEL就是围绕这套协议构建的“管道语法糖”。它最直观的体现就是|运算符——你不需要再嵌套调用，而是像Unix管道一样将数据从一个Runnable流向另一个。

源码/原理深度分析

1. 核心执行引擎：RunnableSequence的调度逻辑

RunnableSequence是LCEL最核心的实现。当你写下runnable1 | runnable2时，LangChain实际创建了一个RunnableSequence对象。我们来看它invoke方法的简化源码（基于LangChain 0.2.x）：

# langchain_core/runnables/base.py
class RunnableSequence(Runnable[Input, Output]):
    def __init__(self, *steps: RunnableLike):
        self.steps = [coerce_to_runnable(step) for step in steps]

    def invoke(self, input: Input, config: Optional[RunnableConfig] = None) -> Output:
        # 关键点1：调用上下文传播
        with ensure_config(config) as config:
            # 关键点2：逐步执行，将第一个步骤的输出作为第二个步骤的输入
            for i, step in enumerate(self.steps):
                if i == 0:
                    output = step.invoke(input, config)
                else:
                    output = step.invoke(output, config)
            return output

这段代码看似简单，但其优雅之处在于配置（config）的透明传递。config中包含了回调处理器（callbacks）、重试策略、元数据等信息。这意味着在链路的任何一步，你都能捕获到完整的执行轨迹。

2. 流式传输的底层魔法：transform与astream

LCEL真正的杀手锏是流式传输。传统的链式调用需要等待前一步全部完成才能进入下一步，而LCEL通过transform方法实现了逐token的流式处理。

# langchain_core/runnables/base.py
class RunnableSequence(Runnable[Input, Output]):
    async def atransform(self, input: AsyncIterator[Input], config: Optional[RunnableConfig] = None):
        # 关键：将输入迭代器传递给第一个步骤
        final_pipeline = self.steps[0].atransform(input, config)
        for step in self.steps[1:]:
            final_pipeline = step.atransform(final_pipeline, config)
        async for chunk in final_pipeline:
            yield chunk

这里的设计思想是惰性求值。每一步都接收一个异步迭代器，并返回一个新的异步迭代器。数据像水流一样，从源头（提示词模板）流出，经过模型生成token，再被解析器逐步处理，全程无需等待完整输出。这就像一条自动化的快递分拣线，包裹（数据块）不必等整卡车货物到达，而是随到随分拣。

3. 并行执行的秘密：RunnableParallel与RunnableBranch

RunnableParallel是LCEL中实现“扇出”的关键。它的invoke方法使用asyncio.gather并发执行所有子任务：

# langchain_core/runnables/base.py
class RunnableParallel(Runnable[Input, Dict[str, Any]]):
    def invoke(self, input: Input, config: Optional[RunnableConfig] = None) -> Dict[str, Any]:
        # 关键：并发执行所有步骤
        with executor as executor:
            futures = {
                key: executor.submit(step.invoke, input, config)
                for key, step in self.steps.items()
            }
            return {key: future.result() for key, future in futures.items()}

而RunnableBranch则实现了基于条件的路由。它的invoke方法会按顺序检查每个条件，一旦匹配就执行对应的分支：

# langchain_core/runnables/branch.py
class RunnableBranch(Runnable[Input, Output]):
    def invoke(self, input: Input, config: Optional[RunnableConfig] = None) -> Output:
        for condition, runnable in self.branches:
            if condition(input):
                return runnable.invoke(input, config)
        # 如果没有匹配，执行默认分支
        return self.default.invoke(input, config)

这种设计让我们可以轻松构建“如果输入包含编程问题，则使用代码专家模型；否则使用通用模型”这类动态路由逻辑。

4. 生命周期钩子：RunnableConfig与回调

LCEL的另一个高级特性是RunnableConfig中的回调机制。它允许你在链路的每一步注入自定义逻辑：

# langchain_core/callbacks/base.py
class BaseCallbackHandler:
    def on_llm_start(self, serialized: Dict[str, Any], prompts: List[str], **kwargs): ...
    def on_llm_new_token(self, token: str, **kwargs): ...
    def on_chain_end(self, output: Output, **kwargs): ...

通过config参数，你可以将回调处理器传递到链路的任意位置，从而实现在模型生成token时实时推送给前端、在链结束时的指标埋点等需求。

实战代码：三个从入门到进阶的完整示例

示例一：构建动态路由的RAG问答链路

这个示例展示如何使用RunnableBranch实现“问题类型检测 + 知识库检索 + 模型生成”的动态链路。

from langchain_core.runnables import RunnableBranch, RunnableLambda, RunnableParallel
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_community.vectorstores import FAISS
from langchain_core.output_parsers import StrOutputParser
from langchain_core.documents import Document

# 初始化组件
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
vectorstore = FAISS.from_documents(
    [Document(page_content="LangChain LCEL 支持流式输出和并行执行。")],
    embedding=OpenAIEmbeddings()
)
retriever = vectorstore.as_retriever()

# 1. 定义问题分类器（返回布尔值）
def is_programming_question(query: str) -> bool:
    keywords = ["code", "python", "bug", "函数", "算法"]
    return any(kw in query.lower() for kw in keywords)

# 2. 定义两条分支链
# 分支A：编程问题 -> 使用检索增强 + 严格模式
programming_prompt = ChatPromptTemplate.from_template(
    """你是资深代码审查专家。基于上下文回答问题。
    
    上下文：{context}
    问题：{question}
    
    要求：给出代码示例，并指出潜在风险。"""
)
programming_chain = (
    RunnableParallel(context=retriever, question=lambda x: x)
    | programming_prompt
    | llm
    | StrOutputParser()
)

# 分支B：通用问题 -> 直接回答
general_prompt = ChatPromptTemplate.from_template("简洁回答问题：{question}")
general_chain = general_prompt | llm | StrOutputParser()

# 3. 构建动态路由
branch = RunnableBranch(
    (is_programming_question, programming_chain),
    general_chain  # 默认分支
)

# 4. 测试
result = branch.invoke("请解释Python中的装饰器原理，并给出示例")
print(result)

关键点：RunnableBranch的第一个参数是条件函数，第二个参数是满足条件时执行的Runnable。这里的RunnableParallel(context=retriever, question=lambda x: x)是LCEL最优雅的用法——它同时执行检索器和透传用户问题，然后将结果合并为字典传给提示词模板。

示例二：流式输出 + 自定义回调监控

这个示例展示如何利用stream方法实现逐token输出，并注入回调监控性能。

from langchain_core.callbacks import BaseCallbackHandler
from langchain_core.runnables import RunnableConfig
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
import time

# 自定义回调处理器
class LatencyCallback(BaseCallbackHandler):
    def __init__(self):
        self.start_time = None
        self.token_count = 0
    
    def on_llm_start(self, serialized, prompts, **kwargs):
        self.start_time = time.time()
        self.token_count = 0
    
    def on_llm_new_token(self, token, **kwargs):
        self.token_count += 1
        # 实时打印token（模拟推送到前端）
        print(f"[Token {self.token_count}]: {token}", end="", flush=True)
    
    def on_llm_end(self, response, **kwargs):
        elapsed = time.time() - self.start_time
        print(f"\n\n=== 完成 ===")
        print(f"生成 {self.token_count} tokens，耗时 {elapsed:.2f}s")
        print(f"吞吐率: {self.token_count / elapsed:.1f} tokens/s")

# 构建带流式输出的链
prompt = ChatPromptTemplate.from_template("写一篇关于{ topic }的500字短文")
llm = ChatOpenAI(model="gpt-4o-mini", streaming=True)
chain = prompt | llm

# 执行流式调用
config = RunnableConfig(callbacks=[LatencyCallback()])
print("开始流式生成：\n")
for chunk in chain.stream({"topic": "人工智能伦理"}, config=config):
    # 这里可以实时处理chunk，比如发送到WebSocket
    pass

关键点：这个例子揭示了LCEL流式调用的重要细节——chain.stream()返回一个生成器，每次迭代产生一个token。通过RunnableConfig传递回调处理器，我们无需修改链路代码即可实现监控、日志、前端推送等横切关注点。

示例三：并行执行多个独立任务并合并结果

这个示例展示如何使用RunnableParallel同时执行摘要、关键词提取和情感分析，然后合并结果。

from langchain_core.runnables import RunnableParallel
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
import asyncio

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# 定义三个独立的子任务
summary_prompt = ChatPromptTemplate.from_template("用3句话总结以下文本：\n{text}")
keywords_prompt = ChatPromptTemplate.from_template("提取文本的5个关键词，用逗号分隔：\n{text}")
sentiment_prompt = ChatPromptTemplate.from_template("判断文本情感（积极/消极/中性）：\n{text}")

summary_chain = summary_prompt | llm | StrOutputParser()
keywords_chain = keywords_prompt | llm | StrOutputParser()
sentiment_chain = sentiment_prompt | llm | StrOutputParser()

# 并行执行
parallel_chain = RunnableParallel(
    summary=summary_chain,
    keywords=keywords_chain,
    sentiment=sentiment_chain
)

# 同步调用
text = "LangChain是一个强大的框架，它通过LCEL让AI应用开发变得前所未有的简单。"
result = parallel_chain.invoke({"text": text})
print(f"摘要: {result['summary']}")
print(f"关键词: {result['keywords']}")
print(f"情感: {result['sentiment']}")

# 异步批量调用（极致性能）
async def async_batch_process(texts):
    results = await parallel_chain.abatch([{"text": t} for t in texts])
    return results

# 测试异步
texts = ["文本1...", "文本2...", "文本3..."]
asyncio.run(async_batch_process(texts))

关键点：RunnableParallel的invoke方法会并发执行所有子链，而不是串行。这里展示的abatch是另一个高级API，它同时利用了异步IO和批量处理，在处理大量文本时性能提升显著。

方案对比：LCEL vs 其他编排方案

在LangChain生态中，LCEL并非唯一的编排方式。下表对比了它与传统Chain、以及新兴LangGraph的差异：

| 特性 | LCEL (Runnable) | 传统 Chain (LLMChain等) | LangGraph |

|------|----------------|------------------------|-----------|

| 组合方式 | \| 管道符，声明式 | 类继承，命令式 | 图结构，显式节点/边 |

| 流式支持 | 一等公民，内置 | 需自定义回调 | 支持，但需配置 |

| 并行 | 内置 RunnableParallel | 需手动 ThreadPool | 通过图拓扑实现 |

| 条件分支 | RunnableBranch | 需 if-else 写在代码中 | 原生支持条件边 |

| 循环/迭代 | 不直接支持 | 不直接支持 | 原生支持（用于代理） |

| 状态管理 | 无（纯函数式） | 无 | 有状态图 |

| 调试友好度 | 配置传递回调，较友好 | 黑盒，较难 | 可视化，最友好 |

| 适合场景 | 简单到中等复杂度的线性/并行链路 | 遗留代码 | 复杂代理、循环、人类介入 |

深度解读：

• LCEL vs 传统Chain：LCEL是LangChain的“现代Hipster”方案，它用组合优于继承的设计哲学彻底替代了Chain类。你可以将LCEL视为Java 8的Stream API取代传统的for循环——功能等价，但表达力、可读性和性能特性完全不同。

• LCEL vs LangGraph：LangGraph更像是LangChain的“图数据库”，它提供了完整的图执行引擎，支持循环、条件边、状态持久化。如果你的链路涉及代理（Agent）的多轮工具调用，LangGraph是更好的选择。但LCEL的简洁性让它成为大多数RAG应用的默认首选。

最佳实践与避坑指南

最佳实践

1. 为每个Runnable命名：使用.with_config({"run_name": "my_step"})，这在追踪日志时能救命。
2. 善用RunnableLambda：当需要插入自定义逻辑时，用RunnableLambda包裹纯函数，而不是强行改造链。
3. 流式优先：在设计链路时，默认考虑stream而非invoke，因为流式调用可以无缝降级为批处理。
4. 组合子最小化：能用|解决的，不要引入RunnableParallel；能用RunnableParallel解决的，不要上LangGraph。

避坑指南

1. 不要用invoke处理大输入：invoke会等所有步骤完成才返回，如果模型生成慢，会导致内存堆积。优先使用stream。
2. 小心RunnableParallel中的共享状态：RunnableParallel的invoke是线程池并发，不要在其中修改外部可变变量，否则会数据竞争。
3. config的显式传递：在自定义Runnable子类中，invoke方法必须显式将config传递给下一步，否则回调、重试等高级特性会静默失效。
4. LLM的streaming必须开启：如果你希望chain.stream()能逐token输出，必须确保底层的ChatOpenAI等模型设置了streaming=True参数。否则，stream()会退化为一次性返回整个结果。
5. 不要混淆batch与parallel：batch是顺序批量处理，RunnableParallel是并行处理。两者在性能特征上完全不同。

总结

LCEL和Runnable协议不仅是一种API设计，更是一种思维模式的转变——从“命令式地编写调用逻辑”转变为“声明式地组合处理单元”。它让AI应用的代码变得可读、可测试、可扩展，同时为流式、并行、动态路由等高级需求提供了第一公民级的支持。

在本文中，我们通过源码分析了RunnableSequence的调度机制、流式传输的惰性求值原理，以及并行和分支的实现细节。三个实战示例覆盖了动态路由、流式监控和并行任务合并三大高频场景。最后，我们对比了LCEL与传统Chain、LangGraph的定位差异，并给出了实践中的关键注意事项。

延伸思考：随着LangChain 0.3+和LangGraph的发展，LCEL正在成为LangGraph中节点的标准接口。这意味着你今天用LCEL构建的每一个Runnable，未来都能无缝嵌入到复杂的图执行流程中。这种“先组合，后编排”的架构演进路径，或许正是AI应用开发走向成熟的重要标志。你的下一个复杂Agent应用，准备好用LCEL来构建了吗？