在构建基于大语言模型（LLM）的应用时，我们经常需要处理Prompt 模板、Chat Model、Output Parser以及Retriever等核心组件。如果每次都通过手动一步步调用这些组件，代码不仅啰嗦，还难以维护和复用。

为了解决这个问题，LangChain 引入了LCEL（LangChain Expression Language）和Runnable协议，让组件组合就像 UNIX 管道符（|）一样优雅直观。

本文将从基础概念出发，通过多个实战案例，手把手带你掌握 LCEL 的核心用法。

1. 什么是 Runnable？

在 LangChain 中，Runnable是所有可执行和可组合组件的基础接口（Protocol）。

我们常用的许多组件本身就是 Runnable 协议的实现：

• PromptTemplate

• ChatModel

• OutputParser

• Retriever

所有 Runnable 组件都继承了一套统一的操作方法：

2. 什么是 LCEL？

LCEL（LangChain Expression Language，LangChain 表达式语言）是 LangChain 提供的一种声明式组合组件的方式。

只要两个组件都实现了Runnable接口，就可以通过管道符|连接成一条执行链（Chain）：

Python

chain = prompt | model | parser

数据会按照管道顺序流动：输入传给 Prompt 模板生成提示词，提示词送入 LLM 模型，模型输出结果再由 Output Parser 进行格式化。

原生 Python vs LCEL 对比

• 传统逐步调用方式：

Python

# 1. 格式化 Prompt prompt_value = prompt.invoke({"topic": "AI"}) # 2. 调用模型 model_output = model.invoke(prompt_value) # 3. 解析输出 result = parser.invoke(model_output)

• 使用 LCEL 管道写法：

Python

chain = prompt | model | parser result = chain.invoke({"topic": "AI"})

LCEL 的优势：

1. 代码极简：结构直观，大幅减少冗余代码。

2. 自动支持高级能力：只要一条 Chain 组装完成，它自动继承invoke、batch、stream以及异步接口。

3. 方便复用：复杂 Chain 可以作为子链嵌套进更大的 Chain 中。

3. LCEL 基础三大调用模式

3.1invoke：单条同步调用

适合用户单次提问、生成单份文档或分析单条数据的场景。

Python

from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser from langchain_openai import ChatOpenAI prompt = ChatPromptTemplate.from_template("请用简明易懂的话解释技术概念：{concept}") model = ChatOpenAI(model="gpt-4o-mini") parser = StrOutputParser() # 组装 Chain chain = prompt | model | parser # 传入字典，单次调用 response = chain.invoke({"concept": "向量数据库"}) print(response)

3.2batch：并发批量处理

当需要对批量数据（如多条评论分类、批量提取简历信息）进行处理时，使用batch能够大幅提升效率。

Python

# 批量传入参数列表 inputs = [ {"product": "无线蓝牙耳机", "feature": "降噪、长续航"}, {"product": "人体工学椅", "feature": "腰靠调节、透气网布"}, {"product": "机械键盘", "feature": "热插拔、三模连接"} ] # 通过 max_concurrency 限制最大并发数，防止触发 Rate Limit results = chain.batch(inputs, config={"max_concurrency": 2}) for item, res in zip(inputs, results): print(f"【{item['product']}】: {res}\n")

提示：通过config={"max_concurrency": N}可以有效限制并发请求数，避免触发 API 频控限制（Too Many Requests）。

3.3stream：打字机流式输出

用于构建 ChatBot 或实时生成长文，实现前端打字机效果。

Python

# 单次调用实现流式输出 for chunk in chain.stream({"concept": "LCEL 表达式语言"}): print(chunk, end="", flush=True)

注意：在终端打印时加上flush=True，可以防止缓冲区堆积导致文本一次性“蹦出”，确保逐字平滑渲染。

4. 进阶组件与复杂链构造

除了标准的 Prompt、Model、Parser，LangChain 还提供了几种特殊的 Runnable 工具，用来处理更复杂的业务逻辑。

4.1RunnableLambda：接入自定义 Python 函数

业务开发中，我们经常需要在送入 Prompt 前对输入数据进行清洗或预处理。使用RunnableLambda可以无缝将普通 Python 函数打包为标准 Runnable。

Python

from langchain_core.runnables import RunnableLambda def clean_input(text: str) -> str: """清除首尾空格并转为小写""" return text.strip().lower() # 将函数包裹为 Runnable clean_step = RunnableLambda(clean_input) # 加入管道：数据先经过 clean_step 清洗，再传递给后续组件 chain = clean_step | prompt | model | parser

4.2RunnableParallel：多任务并行分支

如果需要把同一个输入同时投递给多个处理分支并发执行（例如对一条用户评论同时做“情感分析”、“提取关键词”和“自动回复”），可以使用RunnableParallel。

Python

from langchain_core.runnables import RunnableParallel # 定义三条独立的子链 sentiment_chain = sentiment_prompt | model | parser keywords_chain = keywords_prompt | model | parser reply_chain = reply_prompt | model | parser # 使用 RunnableParallel 并行组合 analysis_pipeline = RunnableParallel({ "sentiment": sentiment_chain, "keywords": keywords_chain, "reply": reply_chain }) # 执行分析 result = analysis_pipeline.invoke({"review": "这款手机屏幕很清晰，但是电池续航太差了！"})

执行机制解析：

• 同一个输入会同时分发给三条子链。

• 内部并发执行，总耗时大约等于最慢的分支耗时，而不是各分支耗时相加。

• 最终返回一个字典，Key 为分支名称，Value 为对应分支的输出结果。

5. RAG 核心神器：RunnablePassthrough

在构建检索增强生成（RAG）管道时，我们常面临一个难题：如何既将用户问题送去检索上下文（context），又将原始问题（question）原封不动传递给 Prompt 模板？

RunnablePassthrough就是专门用来解决数据透传与追加的。

用法 1：原样透传RunnablePassthrough()

当管道遇到{ "key": Runnable... }字典字面量时，LangChain 会自动将其包装为RunnableParallel：

Python

from langchain_core.runnables import RunnablePassthrough # 组合 RAG 链 rag_chain = ( { "context": retriever | format_docs, # 检索知识库并格式化 "question": RunnablePassthrough() # 原样透传用户输入的问题 } | prompt | model | parser ) rag_chain.invoke("什么是 LCEL？")

用法 2：保留原数据并追加新字段.assign()

在实际业务中，我们往往想保留原始字典（如{"question": "..."}），同时新增清洗后的字段或检索到的上下文。使用RunnablePassthrough.assign()可以避免手动繁琐拼接字典：

Python

def clean_question(input_dict: dict) -> str: return input_dict["question"].strip().upper() def mock_retriever(question: str) -> str: return f"关于【{question}】的相关知识库文档内容..." # 使用 assign 追加新字段 processing_chain = RunnablePassthrough.assign( clean_q=RunnableLambda(clean_question), context=lambda d: mock_retriever(d["question"]) ) # 测试执行 output = processing_chain.invoke({"question": " 介绍 assign 用法 "}) print(output)

assign输出结果：

JSON

{ "question": " 介绍 assign 用法 ", // 1. 保留原始字段 "clean_q": "介绍 ASSIGN 用法", // 2. 追加清洗后的新字段 "context": "关于【介绍 assign 用法】的相关知识库文档内容..." // 3. 追加检索结果 }

6. 总结与核心记忆点

掌握 LCEL，是进阶 LangChain 高级开发（如 LangGraph、复杂 RAG 智能体）的必经之路。

建议重点记住以下核心口诀：

1. 一切皆 Runnable：Prompt、Model、Parser、Retriever 都是 Runnable。

2. |连接万物：用管道符组合单链，数据自左向右单向流动。

3. 接口高度统一：单条用invoke，批量用batch，实时用stream。

4. 灵活调度工具：

   • 自定义逻辑用RunnableLambda；

   • 多任务并行用RunnableParallel；

   • 数据透传与字段追加用RunnablePassthrough和assign()。