LangChain LCEL 链

第 2 篇讲提示模板时埋了个伏笔：模板本身就是 Runnable，能用管道符与模型直接串联，| 背后的组装规则是第 5 篇的主场；第 4 篇收尾又预告，见 agent 之前先学链式编排。本篇是《LangChain 从入门到实战》第 5 篇，主角是 LCEL（LangChain Expression Language，LangChain 表达式语言，见官方文档）：先看 Runnable 统一接口如何让所有组件长成同一形状，再用管道符把前四篇的组件串成三行链，然后引入函数、并行、透传三种积木，最后用一条并行双译的翻译流水线收束，并说清 1.x 时代 LCEL 与 agent 框架的分工。

1、Runnable 统一接口

前四篇的组件其实共用着同一套调用协议。第 1 篇调模型的 invoke、第 2 篇渲染模板的 invoke、第 4 篇执行工具的 invoke——不是各组件恰好都起了这个名字，而是它们都实现了 langchain-core 的 Runnable 协议：一组约定的方法与类型，核心三个方法的语义如下：

模板在前四篇只露过 invoke，三件套其实一个不少，示例如下：

from langchain_core.prompts import ChatPromptTemplate

prompt = ChatPromptTemplate.from_messages([
    ("human", "用一句话解释 {term}")
])

prompt.invoke({"term": "管道符"})                     # 渲染一条提示
prompt.batch([{"term": "管道符"}, {"term": "RAG"}])   # 并发渲染两条
list(prompt.stream({"term": "管道符"}))               # 模板不产流，整块产出

解读：同一个对象上三个方法全部可用——模板不联网、渲染极快，batch 的并发在这里感知不强，但接口形状与模型完全一致。这正是 Runnable 设计的用意：接口统一后，调用方不必关心手里拿的是模板、模型还是工具。已出场组件的 Runnable 形状对照如下：

表里五行输入输出两两不同，方法名却完全一致——能在管道里自由拼接的前提，就是这张表。第 4 篇的 bind_tools 返回的绑定模型也在此列：前四篇出场过的调用单元，本篇全部接得上。另外三件套各有一个异步孪生：ainvoke、astream、abatch，Web 服务改写 async 时接口一一对应，本系列示例保持同步写法。

2、管道组合三行成链

先定口径：LCEL 的积木全部来自 langchain-core（当前 1.6.x，随主包安装，见 PyPI 主页）：Runnable 系列在 langchain_core.runnables，字符串解析器在 langchain_core.output_parsers。模型返回的 AIMessage 给程序用时通常只要正文，StrOutputParser 负责剥壳。三行成链的写法如下：

from langchain.chat_models import init_chat_model
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

model = init_chat_model("openai:gpt-6.1-terra", temperature=0)
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是严谨的技术讲师，用中文回答。"),
    ("human", "{question}"),
])

chain = prompt | model | StrOutputParser()
print(chain.invoke({"question": "用一句话说明 LCEL 是什么"}))

| 是 Python 位或运算符的重载（Runnable 实现了 __or__），等价于 prompt.pipe(model).pipe(parser)，习惯方法调用链的写法可以互换。数据在链里的流转如下：

对照第 2、3 篇的管道图，只是把末端延长了一站：字典进、纯文本出，中间的消息形态对调用方完全不可见。两个要点值得点透。其一，链本身也是 Runnable——它由 RunnableSequence 实现，invoke、stream、batch 三件套原样可用，链还能再进管道，组成更大的链。其二，第 3 篇手工写的重试循环在这里一行了结：chain.with_retry(stop_after_attempt=2) 返回带重试的新链，退避与异常筛选参数见 Runnable 接口文档。

链的构造还有一层工程红利：管道是纯声明，拼接那一刻不发任何请求；chain.input_schema 与 chain.output_schema 随时可查，各站的变量、类型在测试里全部可见。第 2 篇说模板可以免 token 进单元测试，这个结论在链上升了一级：整条链的形状都能静态验证，只有真的 invoke 才花钱。

老教程的 LLMChain(llm=..., prompt=...) 在 1.x 主包已不存在——0.x 的 chains 模块整体迁往 langchain-classic。见到这类导入，直接把等价逻辑改写成上面的管道语法。

3、RunnableLambda 入链

管道里的每一站都得是 Runnable，普通函数想入链，用 RunnableLambda 包一层。单独看它就是把函数升级成 Runnable，数据整形是它在实际项目里的主业：

from langchain_core.runnables import RunnableLambda

RunnableLambda(len).invoke("runnable")   # 8，函数立即获得三件套

def with_chars(answer: str) -> dict:     # 给输出补上字数字段
    return {"answer": answer, "chars": len(answer)}

chain2 = prompt | model | StrOutputParser() | RunnableLambda(with_chars)
print(chain2.invoke({"question": "一句话解释管道符"})["chars"])

解读：with_chars 收字符串、吐字典，链的输出形状因此改变——LCEL 不要求各站同构，只要求相邻两站的输出与输入对得上。这个约束与 Unix 管道的精神一致：每站做一件小事，形状对了就能任意拼接。裸函数不建议直接写在管道里，显式包一层 RunnableLambda 意图更清晰，出错时的报错也直指函数本身。RunnableLambda 同样接受异步函数，协程写法在 Web 服务里可以与链的异步三件套无缝对接。

4、并行与上下文注入

两块积木解决"多路分发"与"原样透传"。RunnableParallel 让多个 Runnable 收到同一份输入并发执行，结果按键汇成字典；RunnablePassthrough 把输入原样传出。两者组合最常见的场景是 RAG 的上下文注入——问题与检索结果要同传给提示模板：

from langchain_core.runnables import RunnableLambda, RunnablePassthrough

docs = {"lcel": "LCEL 用管道符组合 Runnable，链本身仍是 Runnable。"}

def retrieve(topic: str) -> str:
    return docs.get(topic, "未找到资料")

rag = (
    {"context": RunnableLambda(retrieve), "question": RunnablePassthrough()}
    | ChatPromptTemplate.from_messages([
        ("system", "只依据参考资料回答，资料不足就明说。"),
        ("human", "参考资料：{context}\n问题：{question}"),
    ])
    | model
    | StrOutputParser()
)

print(rag.invoke("lcel"))   # 输入是裸字符串，问题被透传进模板

三点解读。输入是一个裸字符串，两条支路各取所需：retrieve 拿它查资料，RunnablePassthrough 把它原样放进 question 键——这正是透传的价值：让原始输入与加工产物同框出现。对照裸 SDK 的做法更能看出差别：调用点手工查资料、手工拼 dict、手工塞问题，三件事缠在一处；这里每条支路只声明"输入是什么、输出放哪个键"。字典字面量在管道里会被自动转成 RunnableParallel，与显式构造等价。RunnablePassthrough.assign(key=...) 是它的变体，在透传的同时追加新键，细节见 RunnablePassthrough 参考。把 retrieve 换成真检索器，这段骨架就是完整的 RAG 链——埋线至此，检索器细节留给后面的 RAG 篇。分支路由另有 RunnableBranch 积木，按输入挑一条支路执行，思路与并行同源，篇幅所限不展开。

5、批量与流式

三件套的后两个在链上同样开箱即用。batch 一次提交多个输入，内部并发请求：

questions = [
    {"question": "什么是 Runnable 协议"},
    {"question": "管道符与 pipe 方法等价吗"},
    {"question": "batch 会并发请求吗"},
]

answers = chain.batch(questions, config={"max_concurrency": 3})
print(len(answers))         # 3，顺序与输入列表一致

max_concurrency 控制并发上限，输出顺序与输入严格对应——并发不乱序。对二十条评论跑第 3 篇的分析器，batch 一行替代 for 循环逐个 invoke，总耗时接近单条的量级。流式则贯穿全链，stream 让每站的输出逐段流向下一站：

for chunk in chain.stream({"question": "什么是 LCEL"}):
    print(chunk, end="", flush=True)   # 每段是可直接打印的字符串

模型 SDK 系列讲过裸 SDK 的流式要自己解析分片、手工拼接增量；这里从模型到 StrOutputParser 逐站透传，到手每段已是最终形态的字符串，打印逻辑零拼装。模板这类不支持流式的站也不碍事——它会一次性产出一整块，链整体照样流式。要把链内每一站的进度事件逐级上报时，还有 astream_events 可用，产出带事件类型与站名的结构化日志，服务端监控常用它接进度条。

6、LCEL 与 agent 的分界

代理时代还要不要学 LCEL？官方口径值得单独摆出来。1.x 的文档把 LCEL 定位为轻量的组件编排语言，agent 的构建路径另有其人（以下为大意概括，原文见 LCEL 文档与 agent 文档）：

LCEL 适合用声明式管道把提示、模型、解析器等组件组合成链，轻量、天生支持流式与并发；它不为构建 agent 设计。需要循环、状态与工具编排的 agent，请使用 LangGraph 及其上的 create_agent。

分界线在控制流的结构。LCEL 的管道是无环的：数据从左到右流一遍就结束；agent 的本质是第 4 篇那个"提议、执行、回填"的循环，还要挂状态与断点续跑。两类需求的对比如下：

所以不是"LCEL 过时了"，而是分工收窄了：组件级编排——模板接模型接解析器、并行扇出、函数整形——仍是最顺手的写法，第 3 篇的结构化分析器、本篇的翻译流水线都属此类；循环级编排交给 agent 框架。回到第 1 篇的架构图，langgraph 是编排运行时、主包的 create_agent 是标准 agent 入口，说的正是这条线；LCEL 不在那条线上，它铺在 langchain-core 地基里，是所有组件共享的组装语法——agent 内部的节点拼装用的也是同一套 Runnable 形状。判断口诀一句话：链是声明出来的，agent 是循环出来的。

7、实战翻译流水线

把全部积木组装成一条三段流水线：同一句原文并行跑直译与意译，再由译审合并出终稿。先建两路风格链，风格差异写进各自的系统提示：

def style_prompt(rule: str):
    return ChatPromptTemplate.from_messages([
        ("system", f"你是翻译家。风格要求：{rule}。只输出译文。"),
        ("human", "{source}"),
    ])

literal = style_prompt("直译，忠实原文结构") | model | StrOutputParser()
free = style_prompt("意译，地道自然") | model | StrOutputParser()

并行段把两路译文与随行的原文汇成一个字典，评审段再合成终稿：

def pick_source(payload: dict) -> str:
    return payload["source"]              # 原文随行，评审要用

first = RunnableParallel(
    source=RunnableLambda(pick_source),   # 第 3 节的积木在此复用
    literal=literal,
    free=free,
)

review = (
    ChatPromptTemplate.from_messages([
        ("system", "你是译审。对比两个译文，各取所长合成终稿，只输出终稿。"),
        ("human", "原文：{source}\n直译版：{literal}\n意译版：{free}"),
    ])
    | model
    | StrOutputParser()
)

pipeline = first | review
print(pipeline.invoke({"source": "这代码读起来像诗，跑起来像考古现场。"}))

一次典型运行的输出如下：

This code reads like poetry and runs like an archaeological dig.

整条流水线的结构如下：

四点解读。两路翻译并发执行，总耗时约等于一路——RunnableParallel 的收益直接落在总耗时上。pick_source 是第 3 节 RunnableLambda 的实战复用：并行段的输出必须带上 source 键，评审提示才拿得到原文，上下文要显式随行是并行编排最常见的坑。评审段不过是又一组 prompt | model | parser，三段拼接用的全是本篇讲过的语法——没有新概念，只有旧积木的重排。把 init_chat_model 的字符串换成 anthropic:claude-sonnet-4-5，三段链同时换引擎、结构零改动，统一模型抽象的收益顺着管道传导到了整条流水线。要加第三种风格，在 RunnableParallel 里加一行、评审提示里加一行，改动即完成——声明式组合的扩展成本是线性的。

8、总结与练习

回顾全文：Runnable 协议让模板、模型、解析器、工具长成同一形状，invoke、stream、batch 三件套全组件通用；管道符等价于 pipe 方法，串出的链由 RunnableSequence 实现且仍是 Runnable，with_retry 一行接上重试；RunnableLambda 让普通函数入链做数据整形，RunnableParallel 并发分发、RunnablePassthrough 原样透传，三者拼出 RAG 的上下文注入骨架；官方口径把 LCEL 定位在组件级编排，循环级的 agent 交给 create_agent 与 LangGraph——链是声明出来的，agent 是循环出来的。对于手头的应用，建议把散落的"模板、调用、剥壳"三段代码收成一条三行链，再按需长出并行与函数站；出现循环决策需求的那天，再带着这条链迁移进第 6 篇的 agent。

练习两道：

• 给第 7 节流水线加一条"注释体"支线，数一数需要改动几处
• 用 batch 并发跑十条评论，再换 for 循环逐个 invoke，对比总耗时