主线第四篇里，我们用 prompt | model | parser（Python）/ .pipe()（Node）把几步串成了一条链。很多人会好奇：这个 | 到底做了什么？为什么串起来后就自带 invoke、stream、batch？这一篇专门把 LCEL（LangChain Expression Language）的原理讲透。顶部可切换 Python / Node.js。

一个统一的抽象：Runnable

LCEL 能成立，靠的是一个核心约定：LangChain 里几乎所有组件——模型、提示词模板、输出解析器、检索器、甚至你写的普通函数——都被包装成同一种东西，叫 Runnable（可运行单元）。

只要是 Runnable，就都实现了同一套方法：

• invoke：给一个输入，返回一个输出（最基本）。
• stream：流式返回，一块一块吐。
• batch：一次处理一批输入。
• （Python 还有 ainvoke / astream 等异步版本。）

这就像所有电器都用同一种插头：只要接口统一，就能随意拼接。LCEL 的 | 就是把这些「同款插头」串成一条流水线。

| / .pipe() 到底做了什么

它把左右两个 Runnable 组合成一个新的 Runnable——一个 RunnableSequence（顺序链）。运行时，数据从左流到右：上一个的输出，正好是下一个的输入。

先用最简单的数字例子看清这个「数据流」，不涉及任何模型：

Python

from langchain_core.runnables import RunnableLambda

add1 = RunnableLambda(lambda x: x + 1)   # 把普通函数包装成 Runnable
mul2 = RunnableLambda(lambda x: x * 2)

chain = add1 | mul2                       # 组合成一条顺序链
print(type(chain).__name__)              # RunnableSequence
print(chain.invoke(3))                   # (3+1)*2 = 8

Node.js

import { RunnableLambda } from "@langchain/core/runnables";

const add1 = RunnableLambda.from((x) => x + 1);   // 把普通函数包装成 Runnable
const mul2 = RunnableLambda.from((x) => x * 2);

const chain = add1.pipe(mul2);                    // 组合成一条顺序链
console.log(chain.constructor.name);              // RunnableSequence
console.log(await chain.invoke(3));               // (3+1)*2 = 8

invoke(3) 时：3 进入 add1 得到 4，4 再进入 mul2 得到 8。这就是链的本质——一条传送带，数据在上面被一步步加工。

关键点：因为组合后的 chain 本身也是一个 Runnable，所以它同样有 invoke/stream/batch。这就是为什么你串完 prompt | model | parser 之后，整条链「凭空」就有了流式和批处理能力——不是凭空，是继承自 Runnable 这个统一接口。

Python

print(chain.batch([1, 2, 3]))   # 一次处理一批：[4, 6, 8]
# for chunk in chain.stream(3): ...   # 同样支持流式

Node.js

console.log(await chain.batch([1, 2, 3]));   // 一次处理一批：[4, 6, 8]
// for await (const chunk of await chain.stream(3)) { ... }   // 同样支持流式

两个「自动转换」的贴心设计

你可能注意到主线里有些写法并没有显式包装成 Runnable，却也能进链。那是因为 LCEL 会自动转换两类东西：

1）普通函数 → 自动变成 RunnableLambda。 所以你可以直接把一个函数塞进 |：

Python

chain = add1 | (lambda x: x * 10)   # 右边的函数被自动包装
print(chain.invoke(1))              # (1+1)*10 = 20

Node.js

const chain2 = add1.pipe((x) => x * 10);   // 传入的函数被自动包装
console.log(await chain2.invoke(1));        // (1+1)*10 = 20

2）字典 / 对象 → 自动变成 RunnableParallel（并行链）。 这正是主线第七篇 RAG 里 {"context": retriever | format_docs, "question": RunnablePassthrough()} 的原理：它是一个并行结构，把同一个输入分别喂给每个分支，各自算出结果，组装成一个字典往下传。

Python

from langchain_core.runnables import RunnableParallel

# 同一个输入 5，分别过两个分支，结果合成字典
parallel = RunnableParallel({"a": add1, "b": mul2})
print(parallel.invoke(5))   # {'a': 6, 'b': 10}

Node.js

import { RunnableParallel } from "@langchain/core/runnables";

// 同一个输入 5，分别过两个分支，结果合成字典
const parallel = RunnableParallel.from({ a: add1, b: mul2 });
console.log(await parallel.invoke(5));   // { a: 6, b: 10 }

RunnablePassthrough：原样透传

还有一个常用零件 RunnablePassthrough，作用简单到极致：把输入原封不动地传下去。 它常和并行结构搭配——当你既要拿输入去做某种加工、又想在结果里保留一份原始输入时，就用它占位。

Python

from langchain_core.runnables import RunnablePassthrough

chain = RunnablePassthrough() | add1
print(chain.invoke(4))   # 4 原样穿过，再 +1 → 5

Node.js

import { RunnablePassthrough } from "@langchain/core/runnables";

const chain = new RunnablePassthrough().pipe(add1);
console.log(await chain.invoke(4));   // 4 原样穿过，再 +1 → 5

回看第七篇的 RAG 链就懂了：question 分支用 RunnablePassthrough 把用户问题原样留着填进提示词，而 context 分支把同一个问题拿去检索。两个分支并行，各取所需。

把「玩具」换成真家伙

现在回到主线那条真实的链，你应该能一眼看穿它的数据流了：

Python

# prompt | model | parser 本质就是一条 RunnableSequence
chain = prompt | model | StrOutputParser()
# invoke 时的数据流：
#   {"role":..., "topic":...}
#     → prompt  ：填充变量，输出「一组消息」
#     → model   ：接收消息，输出「AIMessage 对象」
#     → parser  ：接收 AIMessage，输出「纯文本字符串」
text = chain.invoke({"role": "Python 老师", "topic": "什么是装饰器"})

Node.js

// prompt.pipe(model).pipe(parser) 本质就是一条 RunnableSequence
const chain = prompt.pipe(model).pipe(new StringOutputParser());
// invoke 时的数据流：
//   { role, topic }
//     → prompt  ：填充变量，输出「一组消息」
//     → model   ：接收消息，输出「AIMessage 对象」
//     → parser  ：接收 AIMessage，输出「纯文本字符串」
const text = await chain.invoke({ role: "JS 老师", topic: "什么是闭包" });

每一环的输出类型必须能被下一环接收——这就是搭链时唯一要在脑子里对齐的事：上一步吐出来的，下一步能不能吃进去。想不通某条链为什么报错，十有八九就是这里对不上。

为什么要这么设计

统一成 Runnable 带来三个实打实的好处：一是能力免费——任何一条链都自动获得 invoke/stream/batch/异步，不用自己实现；二是可组合——小 Runnable 拼成大 Runnable，大 Runnable 还能再拼，像乐高一样；三是可替换——因为接口一致，把链里某个模型换成另一个、某个检索器换个实现，其余代码一行不用动。

理解了这套「统一接口 + 管道组合」的思想，你再看 LangChain 里任何复杂的链，都能拆解成「一个个 Runnable 按数据流串/并联」——这就是 LCEL 的全部秘密。

---

读完可以回到主线继续：《LangChain 入门》 看它在真实链里的用法，或 《手搓 RAG 问答链》 看并行结构的实战。