在构建基于大语言模型LLM的应用时我们经常需要处理Prompt 模板、Chat Model、Output Parser以及Retriever等核心组件。如果每次都通过手动一步步调用这些组件代码不仅啰嗦还难以维护和复用。为了解决这个问题LangChain 引入了LCELLangChain Expression Language和Runnable协议让组件组合就像 UNIX 管道符|一样优雅直观。本文将从基础概念出发通过多个实战案例手把手带你掌握 LCEL 的核心用法。1. 什么是 Runnable在 LangChain 中Runnable是所有可执行和可组合组件的基础接口Protocol。我们常用的许多组件本身就是 Runnable 协议的实现PromptTemplateChatModelOutputParserRetriever所有 Runnable 组件都继承了一套统一的操作方法方法作用说明invoke处理单条输入最基础的同步调用batch批量处理多条输入并发调用适合批处理任务stream流式返回结果逐字/逐块输出常用于流式打字机效果ainvoke/abatch异步版本适用于 FastAPI 等异步 Web 框架2. 什么是 LCELLCELLangChain Expression LanguageLangChain 表达式语言是 LangChain 提供的一种声明式组合组件的方式。只要两个组件都实现了Runnable接口就可以通过管道符|连接成一条执行链ChainPythonchain 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 管道写法Pythonchain prompt | model | parser result chain.invoke({topic: AI})LCEL 的优势代码极简结构直观大幅减少冗余代码。自动支持高级能力只要一条 Chain 组装完成它自动继承invoke、batch、stream以及异步接口。方便复用复杂 Chain 可以作为子链嵌套进更大的 Chain 中。3. LCEL 基础三大调用模式3.1invoke单条同步调用适合用户单次提问、生成单份文档或分析单条数据的场景。Pythonfrom langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser from langchain_openai import ChatOpenAI prompt ChatPromptTemplate.from_template(请用简明易懂的话解释技术概念{concept}) model ChatOpenAI(modelgpt-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, flushTrue)注意在终端打印时加上flushTrue可以防止缓冲区堆积导致文本一次性“蹦出”确保逐字平滑渲染。4. 进阶组件与复杂链构造除了标准的 Prompt、Model、ParserLangChain 还提供了几种特殊的 Runnable 工具用来处理更复杂的业务逻辑。4.1RunnableLambda接入自定义 Python 函数业务开发中我们经常需要在送入 Prompt 前对输入数据进行清洗或预处理。使用RunnableLambda可以无缝将普通 Python 函数打包为标准 Runnable。Pythonfrom 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 | parser4.2RunnableParallel多任务并行分支如果需要把同一个输入同时投递给多个处理分支并发执行例如对一条用户评论同时做“情感分析”、“提取关键词”和“自动回复”可以使用RunnableParallel。Pythonfrom 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 会自动将其包装为RunnableParallelPythonfrom 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()可以避免手动繁琐拼接字典Pythondef 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_qRunnableLambda(clean_question), contextlambda 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 智能体的必经之路。建议重点记住以下核心口诀一切皆 RunnablePrompt、Model、Parser、Retriever 都是 Runnable。|连接万物用管道符组合单链数据自左向右单向流动。接口高度统一单条用invoke批量用batch实时用stream。灵活调度工具自定义逻辑用RunnableLambda多任务并行用RunnableParallel数据透传与字段追加用RunnablePassthrough和assign()。
【LangChain实战】彻底搞懂 Runnable 与 LCEL：从基础单链到 RAG 复杂管道
在构建基于大语言模型LLM的应用时我们经常需要处理Prompt 模板、Chat Model、Output Parser以及Retriever等核心组件。如果每次都通过手动一步步调用这些组件代码不仅啰嗦还难以维护和复用。为了解决这个问题LangChain 引入了LCELLangChain Expression Language和Runnable协议让组件组合就像 UNIX 管道符|一样优雅直观。本文将从基础概念出发通过多个实战案例手把手带你掌握 LCEL 的核心用法。1. 什么是 Runnable在 LangChain 中Runnable是所有可执行和可组合组件的基础接口Protocol。我们常用的许多组件本身就是 Runnable 协议的实现PromptTemplateChatModelOutputParserRetriever所有 Runnable 组件都继承了一套统一的操作方法方法作用说明invoke处理单条输入最基础的同步调用batch批量处理多条输入并发调用适合批处理任务stream流式返回结果逐字/逐块输出常用于流式打字机效果ainvoke/abatch异步版本适用于 FastAPI 等异步 Web 框架2. 什么是 LCELLCELLangChain Expression LanguageLangChain 表达式语言是 LangChain 提供的一种声明式组合组件的方式。只要两个组件都实现了Runnable接口就可以通过管道符|连接成一条执行链ChainPythonchain 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 管道写法Pythonchain prompt | model | parser result chain.invoke({topic: AI})LCEL 的优势代码极简结构直观大幅减少冗余代码。自动支持高级能力只要一条 Chain 组装完成它自动继承invoke、batch、stream以及异步接口。方便复用复杂 Chain 可以作为子链嵌套进更大的 Chain 中。3. LCEL 基础三大调用模式3.1invoke单条同步调用适合用户单次提问、生成单份文档或分析单条数据的场景。Pythonfrom langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser from langchain_openai import ChatOpenAI prompt ChatPromptTemplate.from_template(请用简明易懂的话解释技术概念{concept}) model ChatOpenAI(modelgpt-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, flushTrue)注意在终端打印时加上flushTrue可以防止缓冲区堆积导致文本一次性“蹦出”确保逐字平滑渲染。4. 进阶组件与复杂链构造除了标准的 Prompt、Model、ParserLangChain 还提供了几种特殊的 Runnable 工具用来处理更复杂的业务逻辑。4.1RunnableLambda接入自定义 Python 函数业务开发中我们经常需要在送入 Prompt 前对输入数据进行清洗或预处理。使用RunnableLambda可以无缝将普通 Python 函数打包为标准 Runnable。Pythonfrom 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 | parser4.2RunnableParallel多任务并行分支如果需要把同一个输入同时投递给多个处理分支并发执行例如对一条用户评论同时做“情感分析”、“提取关键词”和“自动回复”可以使用RunnableParallel。Pythonfrom 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 会自动将其包装为RunnableParallelPythonfrom 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()可以避免手动繁琐拼接字典Pythondef 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_qRunnableLambda(clean_question), contextlambda 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 智能体的必经之路。建议重点记住以下核心口诀一切皆 RunnablePrompt、Model、Parser、Retriever 都是 Runnable。|连接万物用管道符组合单链数据自左向右单向流动。接口高度统一单条用invoke批量用batch实时用stream。灵活调度工具自定义逻辑用RunnableLambda多任务并行用RunnableParallel数据透传与字段追加用RunnablePassthrough和assign()。
