LangChain 指南 · 第 03 章

LCEL 与 Runnable

本章配套可运行示例：demo/demolcel.py。demo 全程用纯数据变换，不启动 Ollama 也能跑；文末留了一段带模型的完整链，想体验的话设 USEMODEL=1 再运行

约 30 分钟11,861 字25 个代码示例

本章配套可运行示例：demo/demo_lcel.py。demo 全程用纯数据变换，不启动 Ollama 也能跑；文末留了一段带模型的完整链，想体验的话设 USE_MODEL=1 再运行。

开场：从「接龙」到「流水线」

前两章我们把两样零件玩熟了：提示词模板负责「把变量填进话术」，模型负责「生成内容」。但「拾光书屋」的真实业务从来不是一个零件能搞定的：

读者提问 → 拼提示词 → 喂模型 → 清洗输出 → 盖上书屋印章 → 展示给读者

如果每一步都手写调用，你的代码会变成一串中间变量的接龙：temp1 = prompt.format(...)、temp2 = model.invoke(temp1)、temp3 = clean(temp2)……步骤一多就晕。

LCEL（LangChain Expression Language）就是终结接龙的：用统一接口把零件串成流水线，拼装一次，反复运行。本章讲透这套拼装术的全部核心积木。

1. LCEL 管道：`|` 与 `.pipe()`

比喻：奶茶店出杯流水线。摇茶机、加料勺、封口机的杯口规格全部统一，所以任何一台都能随意换进产线。| 就是那根传送带。

作用

把多个组件首尾相接：上一个的输出，就是下一个的输入。拼出来的整条链，本身又是一个可运行的组件。

签名（两种等价写法）

chain = first | second | third          # 写法一：| 运算符（Python 惯用，一眼看出流向）
chain = first.pipe(second).pipe(third)  # 写法二：.pipe() 方法（其他语言版本常见写法）

.pipe() 的括号里能塞什么？三类：

Runnable 实例——最常用、最推荐
普通函数 / lambda——自动帮你包成 RunnableLambda（第 3 节）
dict——自动转成 RunnableParallel（第 5 节），键名就是输出字段名

最小示例

纯数据版（不碰模型）：

from langchain_core.runnables import RunnableLambda

# 拾光书屋：书名标准化 -> 盖书屋印章
normalize = RunnableLambda(lambda title: title.strip().title())
stamp = RunnableLambda(lambda title: f"【拾光书屋】{title}")

pipeline = normalize | stamp                    # 等价于 normalize.pipe(stamp)
print(pipeline.invoke("  the little prince "))
# 输出：【拾光书屋】The Little Prince

带模型版（经典三件套）：

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

model = ChatOllama(model="qwen3:4b", temperature=0.7)
prompt = ChatPromptTemplate.from_template("用一句话向社区读者推荐《{book}》")
parser = StrOutputParser()          # 把模型吐出的 AIMessage 变成纯 str

chain = prompt | model | parser     # 模板 -> 模型 -> 解析器
print(chain.invoke({"book": "小王子"}))

数据流向：{"book": "小王子"} →（模板）→ 提示词消息 →（模型）→ AIMessage →（解析器）→ str。

注意一个改名：LangChain 1.x 里字符串解析器叫 StrOutputParser（旧名 StringOutputParser 还能用，但新代码建议写新名）。

常见坑

pipe 的是「机器」，不是「产物」。prompt | model 对，prompt | model.invoke() 错——后者是把一次运行的结果塞上传送带，直接 TypeError。
模板变量和输入 key 要对齐。模板里写 {book}，输入却传 {"title": ...}，校验立刻报错，不会静默通过。
拼装 ≠ 执行。| 只是画好图纸，不调用 invoke / stream / batch，链条一行代码都不会真正跑。

2. Runnable 统一接口：invoke / stream / batch 三兄弟

比喻：不管是螺丝刀还是电钻，开关都是同一个。学会拧这三个开关，你就能驱动 LangChain 世界里的所有组件。

作用

任何 Runnable——单零件也好、拼了几十环的大链也好——都自带同一套运行方法。

签名

result: Output = runnable.invoke(input, config=None)    # 单个输入 -> 完整结果
chunks = runnable.stream(input, config=None)            # 单个输入 -> 逐块产出（生成器）
results: list[Output] = runnable.batch(inputs, config=None)  # 一批输入 -> 一批结果

最小示例

from langchain_core.runnables import RunnableLambda

price_of = RunnableLambda(lambda b: f"《{b['title']}》定价 {b['price']} 元")

# 1) invoke：一次一个
print(price_of.invoke({"title": "小王子", "price": 39}))
# 《小王子》定价 39 元

# 2) stream：拆成一块块吐出（纯数据链通常一次只吐一块）
for chunk in price_of.stream({"title": "夜航", "price": 28}):
    print("收到一块：", chunk)

# 3) batch：一次一排，结果顺序与输入一一对应
cart = [
    {"title": "小王子", "price": 39},
    {"title": "夜航", "price": 28},
    {"title": "人类群星闪耀时", "price": 45},
]
for i, r in enumerate(price_of.batch(cart), start=1):
    print(f"第{i}条：", r)

输入输出类型检查

链是首尾相接的，每一环的输出必须能被下一环吃下。心里要有一张「数据形状表」：

环节	吃什么	吐什么
ChatPromptTemplate	dict（key 与模板变量对上）	提示词消息（ChatPromptValue）
ChatOllama	提示词消息 / 消息列表	AIMessage
StrOutputParser	消息对象	str
RunnableLambda	你说了算	你说了算

想偷看某个组件的「接口形状」，查它的 schema：

from langchain_core.prompts import ChatPromptTemplate

prompt = ChatPromptTemplate.from_template("给《{book}》写一句推荐语")
print(prompt.input_schema.model_json_schema())
# {'properties': {'book': {'title': 'Book', 'type': 'string'}}, 'required': ['book'], ...}

给 RunnableLambda 包的函数加上类型标注，schema 会跟着变准——这也是我们坚持写标注的原因：

from pydantic import BaseModel
from langchain_core.runnables import RunnableLambda

class Book(BaseModel):
    title: str
    price: float

def discount(b: Book) -> str:
    return f"会员价：{b.price * 0.9:.2f} 元"

checker = RunnableLambda(discount)
print(checker.input_schema.model_json_schema())   # 能看到 title / price 两个字段
print(checker.invoke(Book(title="夜航", price=28)))  # 会员价：25.20 元

类型对不上时，报错往往发生在下游消费数据的那一刻：函数里写 x["title"] 却传了个字符串进来，会得到 TypeError: 'str' object is not subscriptable。排查思路：沿着链一环一环单独 invoke，看断在哪一环。

常见坑

batch 和 stream 没法一次同时要：batch 的范式是「一批完整结果」，stream 的范式是「单条输入的碎片流」，两者返回结构不同，一次调用只能选一种。想要「批量 + 流式」，得自己控制并发——第 04 章专门讲这个。
batch 的输出顺序与输入顺序一一对应，不用担心乱序。
对不支持真正流式的组件调 stream 不会报错，只是退化成「一次性吐一大块」。

3. RunnableLambda：给普通函数发上岗证

比喻：转接头。普通函数是个两脚裸插头，插不进流水线的三孔插座；RunnableLambda 给它套个壳，立刻变成标准件。

作用

把任意普通函数（含 async 函数）包装成 Runnable，从而能进管道、能被 invoke / stream / batch 驱动、能和模板模型混拼。

签名

from langchain_core.runnables import RunnableLambda

runnable_fn = RunnableLambda(func)   # 直接传函数本体

最小示例

from langchain_core.runnables import RunnableLambda

def highlight(text: str) -> str:
    """给书名关键词加高亮记号"""
    return text.replace("小王子", "★小王子★")

embellish = RunnableLambda(highlight)
print(embellish.invoke("本月共读书目是《小王子》"))
# 本月共读书目是《★小王子★》

接在模型链后面做后处理：

# chain = prompt | model | parser | RunnableLambda(highlight)

经典用途：

输出后处理：高亮、脱敏、格式清洗
插日志、埋点
调外部 API：翻译、情感分析、敏感词过滤
条件中断：输入不合规就提前抛错，拦住后面的昂贵调用

常见坑

从其他语言版本迁移过来的同学注意：Python 里是 RunnableLambda(fn)，没有 .from() 这种静态方法。
函数只吃一个输入，这个输入通常是整个 dict。想要多个参数？传 dict 进去，函数里自己解包。
函数忘了 return，下游拿到 None——错误会在很远的下游才爆出来。这一环写完先单独 invoke 一下自查。

4. RunnablePassthrough：原样放行与顺手塞便签

比喻：过桥不留客。RunnablePassthrough 就是一段直管道，什么进来什么出去。单看似乎没用，但配上 .assign()，它就变成「过桥时顺手往包裹里塞便签」的组件——原数据一个不动，新字段并行追加。

作用

直通：输入原样传出。常用来占位、对齐链的结构。
assign：保留原有全部字段，并行计算若干新字段并追加进去。

签名

from langchain_core.runnables import RunnablePassthrough

RunnablePassthrough()                     # 直通
RunnablePassthrough.assign(**kwargs)      # 追加字段，值可以是函数 / Runnable

最小示例

from langchain_core.runnables import RunnablePassthrough

# 直通：进什么，出什么
print(RunnablePassthrough().invoke("拾光书屋"))
# 拾光书屋

# assign：原字段保留，新字段并行追加
contextualize = RunnablePassthrough.assign(
    community=lambda _: "拾光书屋",
    member_level=lambda q: "黄金会员" if len(q["query"]) > 10 else "普通会员",
)
print(contextualize.invoke({"query": "推荐几本关于星际旅行的书"}))
# {'query': '推荐几本关于星际旅行的书',
#  'community': '拾光书屋',
#  'member_level': '黄金会员'}

真实项目里最常见的用法：把「检索结果」assign 进输入，再交给模板——第 08 章 RAG 会反复用这个套路：

# retrieve_docs 是一个 Runnable（比如向量检索器）
# chain = RunnablePassthrough.assign(docs=retrieve_docs) | prompt | model | parser

常见坑

assign 的输入必须是 dict，传字符串进来会得到 ValueError: The input to RunnablePassthrough.assign() must be a dict.
assign 里的值函数拿到的也是整个输入 dict，不是某个字段。
新字段与旧字段同名时，新字段直接覆盖。
从其他语言版本迁移的同学注意：那边有 new RunnablePassthrough({ func }) 这种「直通 + 副作用」玩法，Python 版没有对应参数——想加日志之类的副作用，用 RunnableLambda 打一行日志再 return x 原样返回即可。

5. RunnableParallel：一变多的分身术

比喻：复印机。一页原稿（输入）复印几份，同时发给不同部门（分支）处理，最后各部门把结果装进同一个档案袋（dict）交回来。

作用

多个分支拿到同一份输入并发执行，结果按键名汇总成一个 dict 返回。

签名

from langchain_core.runnables import RunnableParallel

fan_out = RunnableParallel(
    summary=chain_a,     # 键名 = 输出字段名，值可以是函数 / Runnable / 子链
    keywords=chain_b,
)

偷懒写法：直接把 dict 交给 | 或 .pipe()，会被自动转成 RunnableParallel，效果完全一样：

chain = some_runnable | {"summary": chain_a, "keywords": chain_b}

最小示例

纯数据版：

from langchain_core.runnables import RunnableLambda, RunnableParallel

stats = RunnableParallel(
    title_len=lambda b: len(b["title"]),
    is_long=lambda b: b["pages"] > 300,
)
print(stats.invoke({"title": "时间简史", "pages": 268}))
# {'title_len': 4, 'is_long': False}

带模型版——一条输入同时要「摘要」和「关键词」，两路并发，总耗时约等于较慢的那一路：

summary_chain = summary_prompt | model | parser
keywords_chain = keywords_prompt | model | parser

fan_out = RunnableParallel(summary=summary_chain, keywords=keywords_chain)
print(fan_out.invoke({"topic": "星际旅行"}))
# {'summary': '……', 'keywords': '……'}

顺带说一句：RunnableMap 是 RunnableParallel 的别名，两个名字指向同一个类。你在别的资料里看到 RunnableMap，搬到 Python 里直接写 RunnableParallel 就行。

常见坑

每个分支拿到的是同一份完整输入，不是把输入切成片分发。想分片请用 batch。
输出是 dict，键名就是你在 parallel 里写的名字——下一环（通常是模板）的变量名要和它对齐。
分支之间别有先后依赖。真有依赖就别硬并行，老老实实用 | 串行。

6. 条件路由：流水线上的分诊台

比喻：医院分诊台。先问诊（判别），再决定挂哪个科（选择），最后进科室看病（执行）。条件路由让系统不再押注单一路径，把「该走哪条路」本身设计成流水线的一环，避免「一个提示词走天下」。

路由条件既可以是规则驱动（关键字、正则、用户等级、时间段），也可以是模型辅助（先让轻量模型判断意图，再分流）。命中不了要有兜底。

作用

根据输入内容把请求分发到不同的子链。LangChain 提供两种姿势：

RunnableBranch：声明式路由表，从上往下匹配，命中即执行，最后必须给一个默认分支。
逻辑分支（官方更推荐）：在普通函数里写 if/else，自己决定走哪条子链。

签名

from langchain_core.runnables import RunnableBranch

branch = RunnableBranch(
    (条件1, 子链1),    # 每项是一个 (条件, 子链) 元组
    (条件2, 子链2),
    默认子链,           # 最后一项不带元组，作为兜底，不能省
)

最小示例

场景：拾光书屋客服台，读者提问分流到「订单」「荐书」「闲聊」三条支线。

写法一，RunnableBranch：

from langchain_core.runnables import RunnableBranch, RunnableLambda

desk = RunnableBranch(
    (
        lambda q: "订单" in q["question"],
        RunnableLambda(lambda q: f"[订单台] 查询：{q['question']}"),
    ),
    (
        lambda q: any(w in q["question"] for w in ("推荐", "找书", "书单")),
        RunnableLambda(lambda q: f"[荐书台] 荐书：{q['question']}"),
    ),
    RunnableLambda(lambda q: f"[前台] 闲聊：{q['question']}"),   # 兜底
)

print(desk.invoke({"question": "我的订单到哪了"}))      # [订单台] 查询：我的订单到哪了
print(desk.invoke({"question": "推荐几本科幻小说"}))    # [荐书台] 荐书：推荐几本科幻小说
print(desk.invoke({"question": "今天天气不错"}))        # [前台] 闲聊：今天天气不错

写法二，逻辑分支——把「判别 + 选择 + 执行」收进一个函数，条件想写多复杂都行：

from langchain_core.runnables import RunnableLambda

def triage(q: dict) -> str:
    text = q["question"]
    if "订单" in text:
        return f"[订单台] 查询：{text}"
    if any(w in text for w in ("推荐", "找书", "书单")):
        return f"[荐书台] 荐书：{text}"
    return f"[前台] 闲聊：{text}"

triage_runnable = RunnableLambda(triage)
print(triage_runnable.invoke({"question": "我的订单到哪了"}))

带模型的进阶玩法（第 10、11 章会大量出现）：判别这一步也交给模型——先用一条「分类链」打标签，再按标签选子链：

# classify_chain = 分类提示词 | model | parser      # 输出「荐书 / 订单 / 闲聊」之一
# def router(q: dict) -> str:
#     label = classify_chain.invoke(q)
#     chain = {"荐书": book_chain, "订单": order_chain}.get(label, chat_chain)
#     return chain.invoke(q)
#
# final_chain = RunnableLambda(router)

两种姿势怎么选：

分支少、条件简单、想让路由表一眼看清 → RunnableBranch
条件复杂、要动态挑链、要和外部系统联动 → 逻辑分支

常见坑

Python 版 RunnableBranch 的分支是(条件, 子链) 元组，最后一项是不带元组的默认分支——构造时就要求至少两个分支（一条条件分支 + 兜底），少了直接 ValueError。
条件从上到下短路匹配：上面的条件写得太宽，下面的分支永远轮不到。
条件函数的返回值只要「真值」即可生效，但强烈建议规规矩矩返回 bool，避免字符串、数字带来的「意外为真」。

7. 链的嵌套组合：拼好的积木继续当零件

比喻：乐高。几块小积木拼成一个「部件」，这个部件本身还能继续拼进更大的作品。链也一样——拼好的链是个 Runnable，天然可以继续 |。

作用

把链当作更高层链的零件，搭出多层结构。RunnableSequence 则提供「显式声明步骤列表」的写法，步骤顺序一目了然，适合步骤较多的正式场景。

签名

from langchain_core.runnables import RunnableSequence

chain = RunnableSequence(step1, step2, step3)   # 位置参数依次传入

最小示例

from langchain_core.runnables import (
    RunnableLambda,
    RunnableParallel,
    RunnablePassthrough,
    RunnableSequence,
)

# 基础零件
clean = RunnableLambda(lambda q: {**q, "query": q["query"].strip()})
enrich = RunnablePassthrough.assign(scope=lambda _: "科幻")
fanout = RunnableParallel(
    recommend=lambda q: f"荐书结果：{q['query']}",
    hot=lambda _: "本周热门：《三体》",
)

# 方式一：继续用 | 拼装
pipeline_a = clean | enrich | fanout

# 方式二：RunnableSequence 显式声明步骤
pipeline_b = RunnableSequence(clean, enrich, fanout)

print(pipeline_a.invoke({"query": "  星际旅行  "}))
print(pipeline_b.invoke({"query": "  星际旅行  "}))
# 两者输出相同：
# {'recommend': '荐书结果：星际旅行', 'hot': '本周热门：《三体》'}

带模型版的完整流水线（示意）：

preprocess = RunnablePassthrough.assign(scope=lambda _: "科幻")
qa_chain = qa_prompt | model | parser
full = RunnableSequence(preprocess, qa_chain, RunnableLambda(save_to_log))

常见坑

从其他语言版本迁移的同学注意：那边常见的 RunnableSequence.from([a, b, c])（传数组）习惯要改——Python 版直接传位置参数，把 list 传进去会报 TypeError: Expected a Runnable, callable or dict。
用 | 拼出来的链本质上就是 RunnableSequence，两种写法可以混用、结果一致。
嵌套别太深。层数一多，中间数据的形状最好用类型标注或注释交代清楚，不然三个月后的自己都读不懂。

8. Runnable 到底统一了什么

比喻：会员卡。拾光书屋的会员卡一张通用：借书、听讲座、买咖啡都刷它。Runnable 就是 LangChain 世界的通用会员卡——只要实现了这套接口，就能刷进任何流水线。

本章出场过的所有角色，底层全是 Runnable：

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import (
    Runnable,
    RunnableLambda,
    RunnablePassthrough,
    RunnableParallel,
)
from langchain_ollama import ChatOllama

members = [
    ChatPromptTemplate.from_template("推荐《{book}》"),   # 模板
    ChatOllama(model="qwen3:4b"),                         # 模型
    StrOutputParser(),                                    # 解析器
    RunnableLambda(lambda x: x),                          # 函数包装
    RunnablePassthrough(),                                # 直通
    RunnableParallel(a=lambda x: x),                      # 并行
]
for m in members:
    print(f"{type(m).__name__:20s} isinstance(m, Runnable) -> {isinstance(m, Runnable)}")
# 六行全是 True

统一带来三个实打实的好处：

拼装自由：零件之间接口一致，任何两个都能用 | 相连，不用为「谁接谁」写胶水代码。
运行一致：invoke / stream / batch 对所有零件通用，三个方法学到手，整座工具箱都会用。
无限组合：链是 Runnable，链的链还是 Runnable——小链拼大链，大链拼系统，理论上可以一直套娃。

这也是 LCEL 的设计初衷：与其给每种组件各设计一套调用方式，不如约定一个统一接口，让「组合」这件事变得零成本。

小结

一张速查表带走本章：

组件 / 写法	一句话定位	典型场景
`a \\| b` / `a.pipe(b)`	串行拼装	模板 → 模型 → 解析器
invoke / stream / batch	统一运行入口	单条 / 流式 / 批量
RunnableLambda	普通函数变标准件	后处理、埋点、路由
RunnablePassthrough	原样直通 + assign 补字段	注入检索结果、上下文
RunnableParallel	同一输入并发分身	摘要 + 关键词同时算
RunnableBranch	声明式条件路由	意图分流、风控兜底
RunnableSequence	显式步骤列表	多层流水线

拼装术练完了，下一章把 stream 和 batch 这对能力展开讲透：流式输出怎么做打字机效果、批量请求怎么控制并发上限——LCEL 流水线的「驾驶技巧」全在下一章。

下一章：04 · 流式与并发

开场：从「接龙」到「流水线」

1. LCEL 管道：| 与 .pipe()

作用

签名（两种等价写法）

最小示例

常见坑

2. Runnable 统一接口：invoke / stream / batch 三兄弟

作用

签名

最小示例

输入输出类型检查

常见坑

3. RunnableLambda：给普通函数发上岗证

作用

签名

最小示例

常见坑

4. RunnablePassthrough：原样放行与顺手塞便签

作用

签名

最小示例

常见坑

5. RunnableParallel：一变多的分身术

作用

签名

最小示例

常见坑

6. 条件路由：流水线上的分诊台

作用

签名

最小示例

常见坑

7. 链的嵌套组合：拼好的积木继续当零件

作用

签名

最小示例

常见坑

8. Runnable 到底统一了什么

小结

1. LCEL 管道：`|` 与 `.pipe()`