前置衔接：前面我们学了 Prompt 模板规范输入、OutputParser 规范输出，本篇带你用 LCEL 表达式 把所有组件串成标准化流水线，彻底告别臃肿的嵌套调用代码。

---

一、前言：新手写 AI 应用的通病

很多同学刚接触 LangChain 时，写代码都是一步步嵌套调用：

# 新手典型写法：层层嵌套，可读性极差
prompt_text = prompt_template.format(name="鼠标", price=99)
messages = ChatPromptTemplate.from_template(prompt_text).invoke({})
response = model.invoke(messages)
result = parser.invoke(response)

这种写法写短链还好，一旦业务复杂起来：要加检索、加数据处理、加分支判断、加并行调用，代码会迅速膨胀成“意大利面”，改一个环节要翻半天。

而且还会遇到一堆头疼问题：

• 想加流式输出，每个组件都要改一遍代码
• 想批量跑多个输入，得自己写循环 + 并发
• 想调试看每一步输入输出，得手动打日志
• 替换组件（换模型 / 换解析器）牵一发而动全身

LCEL 就是专门解决这个问题的官方标准答案。

---

二、大白话讲懂：Runnable 与 LCEL 到底是什么？

2.1 Runnable：统一的“可运行单元”

在 LCEL 的世界里，万物皆 Runnable。

提示词模板、大模型、输出解析器、检索器、甚至你自己写的函数，只要实现了 Runnable 接口，就都是标准化的“工位”，都有统一的调用方法：

• .invoke()：单次执行，拿到最终结果
• .stream()：流式输出，逐 token 返回
• .batch()：批量并发执行多个输入
• .ainvoke()：异步版本

就像所有 USB 设备都能插 USB 接口一样，所有 Runnable 都能用同一套规则调用、同一套方式组合。

2.2 LCEL：用管道把组件串成流水线

LCEL = LangChain Expression Language（LangChain 表达式语言）

它最核心的设计就是用管道符 | 连接各个 Runnable 组件，左边的输出自动作为右边的输入，一行代码就能拼出完整业务链。

类比一下：

• 传统写法 = 手工搬运物料，每个工位之间你自己传数据
• LCEL 写法 = 搭好一条传送带，物料自动顺着流水线往后走

# 传统嵌套写法
result = parser.invoke(model.invoke(prompt.invoke(data)))
LCEL 管道写法
chain = prompt | model | parser
result = chain.invoke(data)

代码逻辑一目了然，哪个环节要替换，直接增删管道里的组件就行。

---

三、LCEL 到底强在哪？四大核心优势

能力	传统手写调用	LCEL 表达式
代码可读性	嵌套层级深，逻辑难梳理	管道线性结构，一眼看懂流程
流式输出	每层手动处理流，代码冗余	一行 .stream() 全链路自动流式
批量并发	自己写循环 + 多线程，容易出错	内置 .batch()，自动并发执行
调试追踪	手动打日志，排查困难	原生对接 LangSmith，全链路可视化
组件替换	修改多处代码，易出 bug	管道里插拔替换，零侵入

💡 本质上 LCEL 不是什么黑科技，它是一套 标准化的组件组合规范。你不用再关心数据怎么在组件间传递，不用自己处理流式、异步、并发，专注写业务逻辑就行。

---

四、五大核心 Runnable 组件实战

4.1 RunnableSequence：串行管道（最常用）

也就是 | 管道符的底层实现，组件按顺序依次执行，前一个的输出是后一个的输入，90% 的基础场景都用它。

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from utils.model_factory import get_deepSeek_model
初始化组件
model = get_deepSeek_model(0.7)
prompt = ChatPromptTemplate.from_messages([
("system", "你是专业电商文案师，输出精简有力"),
("human", "给商品{product}写一句15字以内的宣传语")
])
parser = StrOutputParser()
用管道串成链（RunnableSequence）
chain = prompt | model | parser
统一调用
result = chain.invoke({"product": "无线静音鼠标"})
print(result)

只要链拼好了，换流式输出只改一个方法：

for token in chain.stream({"product": "无线静音鼠标"}):
    print(token, end="", flush=True)

4.2 RunnableParallel：并行执行（提升性能）

同一个输入，同时走多个分支处理，最后合并成字典返回。适合多路检索、多模型对比、同时生成多份内容的场景。

from langchain_core.runnables import RunnableParallel
定义两个分支链
chain1 = ChatPromptTemplate.from_template("解释{topic}的定义，30字以内") | model | parser
chain2 = ChatPromptTemplate.from_template("列出{topic}的3个核心特点") | model | parser
并行执行两个链
parallel_chain = RunnableParallel(
definition=chain1,
features=chain2
)
result = parallel_chain.invoke({"topic": "LCEL表达式"})
print("定义：", result["definition"])
print("特点：", result["features"])

两个分支会 同时并发执行，总耗时不是两者相加，而是取耗时更长的那个，性能提升非常明显。

4.3 RunnableLambda：自定义函数塞进链里

业务里总有一些自定义逻辑：清洗文本、计算长度、格式化数据，用 RunnableLambda 就能把普通函数包装成 Runnable，无缝插进管道里。

from langchain_core.runnables import RunnableLambda
自定义处理函数：输入必须是单个参数
def text_clean(text: str) -> str:
"""清洗模型输出，去除多余空格换行"""
return text.strip().replace("\n", " ")
包装成 Runnable
clean_func = RunnableLambda(text_clean)
插进管道里：模型输出 → 清洗 → 返回
chain = prompt | model | parser | clean_func
result = chain.invoke({"product": "机械键盘"})
print(result)

注意：自定义函数只能接收 一个入参。如果要传多个参数，就用字典打包进去，函数里自己解包。

4.4 RunnablePassthrough：数据透传与字段扩展

非常实用的组件，作用是 把输入原样传递下去，同时新增字段，是复杂链里做数据传递的核心。

最常用的是 .assign() 方法：保留原有所有字段，新增计算字段。

from langchain_core.runnables import RunnablePassthrough
原始输入透传，同时新增 text_length 字段
chain = RunnablePassthrough.assign(
text_length=lambda x: len(x["content"])
)
result = chain.invoke({"content": "这是一段测试文本"})
print(result)  # 输出：{'content': '这是一段测试文本', 'text_length': 8}

典型场景：RAG 检索链里，用户问题透传给 Prompt，同时把检索到的上下文也塞进去。

4.5 输出解析器：结构化输出的最后一环

也就是我们上一篇学的 OutputParser，本身就是 Runnable，直接接在管道末尾，自动把模型输出转成字典 / 对象。

from langchain_core.output_parsers import JsonOutputParser
json_parser = JsonOutputParser()
chain = prompt | model | json_parser
result = chain.invoke({"product": "蓝牙耳机"})
print(type(result))  # dict

---

五、完整实战：搭建一个商品信息生成链

把上面的组件组合起来，做一个真实业务场景：输入商品名称，并行生成文案 + 卖点，最后统一整理成结构化 JSON。

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnableParallel, RunnablePassthrough
from langchain_core.output_parsers import JsonOutputParser
from utils.model_factory import get_deepSeek_model
model = get_deepSeek_model(0.5)
parser = JsonOutputParser()
1. 两个生成分支
copy_chain = ChatPromptTemplate.from_template(
"给{product}写30字以内商品文案，突出使用体验"
) | model | StrOutputParser()
feature_chain = ChatPromptTemplate.from_template(
"列出{product}的3个核心卖点，用顿号分隔"
) | model | StrOutputParser()
2. 并行生成，同时透传原始商品名
parallel_step = RunnableParallel(
product=RunnablePassthrough(),
copywriting=copy_chain,
selling_points=feature_chain
)
3. 最终整理成结构化 JSON
format_prompt = ChatPromptTemplate.from_template("""
根据以下信息整理成 JSON，包含字段：商品名称、宣传文案、核心卖点
商品名称：{product}
宣传文案：{copywriting}
核心卖点：{selling_points}
只输出 JSON，不要多余内容
""")
final_chain = parallel_step | format_prompt | model | parser
4. 执行调用
result = final_chain.invoke("无线静音鼠标")
print(result)

整个链路的执行流程非常清晰：

输入商品名 → 并行生成文案+卖点 → 汇总格式化 → 输出 JSON

后续要加环节、换模型、改逻辑，都只要在对应位置插拔组件即可。

---

六、新手高频踩坑避坑指南

6.1 管道两边的输入输出格式不匹配

坑点：左边输出是字符串，右边组件需要字典，直接报错。

解决：中间加 RunnableLambda 做格式转换，确保前后组件的数据格式对齐。

6.2 普通函数直接塞进管道

坑点：直接写 prompt | model | my_func 会报错，普通函数不是 Runnable。

解决：用 RunnableLambda(my_func) 包装一层再接入。

6.3 并行分支 key 名冲突

坑点：RunnableParallel 里多个分支输出同名字段，后面的会覆盖前面的。

解决：给每个分支起唯一的 key 名，接收时按 key 取值。

6.4 认为 LCEL 只能用 LangChain 自带组件

误区：不是只有官方组件才能用管道。任何逻辑只要包装成 Runnable，都能无缝接入，自由度非常高。

---

七、全文核心总结

1. Runnable 是 LangChain 的统一接口，所有组件都遵循同一套调用规范，支持 invoke/stream/batch/异步四大执行方式。
2. LCEL 是声明式的管道语法，用 | 串联组件，让数据自动流转，代码更简洁、易维护。
3. 五大核心组件各司其职：
   • RunnableSequence（）：串行执行，基础流水线
   • RunnableParallel：并行分支，提升并发性能
   • RunnableLambda：自定义逻辑，无缝入链
   • RunnablePassthrough：数据透传与字段扩展
   • OutputParser：收尾做结构化输出
4. LCEL 不是炫技语法，它的核心价值是 工程化：统一标准、降低维护成本、原生支持流式 / 并发 / 调试，是企业级项目的标准写法。

---

如果文章对你有帮助，点赞 + 收藏 + 关注，持续更新 LangChain 从零到企业级落地全套实战教程！