LCEL 是什么——声明式组合的哲学在早期版本的 LangChain 中构建一条处理链需要命令式地依次调用每个组件先调用 Prompt 生成消息再把消息传给 LLM 得到响应最后调用 Parser 解析输出。这种写法直观但随着链路变长嵌套调用和中间变量会迅速积累代码可读性和可维护性都会下降。更麻烦的是流式输出、批量处理、异步执行这些能力无法自动复用每条链都需要单独处理。LCELLangChain Expression Language用声明式的思路解决了这个问题。你只需要描述这些组件按什么顺序组合而不需要关心每一步如何传递数据。LCEL 的核心思想是组合本身就是值你用|把几个组件串在一起得到的是一个新的可执行对象流式、批量、异步等能力由框架统一提供无需重复实现。这与函数式编程中的管道pipeline概念一脉相承让链的结构一眼就能看清楚。Runnable万物皆可组合的统一接口LCEL 能够把不同类型的组件自由组合依赖的是一个统一的抽象Runnable接口。LangChain 中几乎所有的核心组件等都实现了这个接口因此它们在 LCEL 眼中是等价的积木可以任意拼接。Runnable对外暴露三种主要的调用方式分别对应不同的使用场景。invoke是最基础的同步调用接收一个输入等待整个链执行完毕后返回最终结果适合单次请求。stream则以迭代器的方式逐块返回结果LLM 每生成一个 token 就立刻推送出来用户不需要等到全部生成完才看到响应非常适合对话类应用。batch接收一个输入列表在内部并发执行多个invoke比顺序循环调用效率高得多适合需要批量处理文本的离线任务。三种方式共用同一条 chain 定义切换成本极低。下面的示例以解释一个概念的简单链为例依次演示这三种调用fromdotenvimportload_dotenvfromlangchain_core.promptsimportChatPromptTemplatefromlangchain_core.output_parsersimportStrOutputParserfromlangchain_openaiimportChatOpenAI# 读取 .env 文件并注入环境变量之后 LangChain 会自动从环境变量中获取密钥load_dotenv()promptChatPromptTemplate.from_template(用一句话解释{concept})llmChatOpenAI(modelQwen/Qwen3.6-35B-A3B,temperature0,base_urlhttps://api.siliconflow.cn/v1)parserStrOutputParser()chainprompt|llm|parser# invoke同步等待完整结果resultchain.invoke({concept:递归})print(result)# stream逐 token 打印无需等待全部完成forchunkinchain.stream({concept:递归}):print(chunk,end,flushTrue)# batch并发处理多个输入resultschain.batch([{concept:递归},{concept:闭包},{concept:协程}])print(results)这段代码先用prompt | llm | parser定义了一条简单的三步骤链。invoke是最常见的调用方式传入一个字典返回最终文本整个过程是同步阻塞的。stream则把链中的 LLM 调用切换为流式输出模式每次只返回一个 token用for循环边接收边打印用户能实时看到文字逐字出现。batch接收的是一个列表内部对列表中的每个元素发起一次独立的invoke并且这些调用是并发执行的三项概念的解释会在大致相同的时间内完成而不是按顺序逐个等待。管道运算符|的秘密|在 Python 里是位运算的按位或运算符但 LangChain 通过重载__or__和__ror__魔术方法赋予了它全新的含义。当你写step1 | step2时Python 实际上调用的是step1.__or__(step2)而Runnable的这个方法会返回一个RunnableSequence对象把两个步骤封装在一起。关键在于|只是声明不是执行。chain prompt | llm | parser这行代码执行完之后没有任何网络请求发出没有任何 token 被生成chain只是一个描述先走 prompt再走 llm最后走 parser的配置对象。真正的执行发生在你调用.invoke()、.stream()或.batch()的时候。这种惰性设计让你可以把链的构建和链的执行完全分离也让链本身可以被序列化、复用和传递。下面用一条简单的翻译链来验证这一点注意代码中注释标注的声明与执行两个阶段fromdotenvimportload_dotenvfromlangchain_core.promptsimportChatPromptTemplatefromlangchain_core.output_parsersimportStrOutputParserfromlangchain_openaiimportChatOpenAI# 读取 .env 文件并注入环境变量之后 LangChain 会自动从环境变量中获取密钥load_dotenv()# 声明三步串联的链——此刻什么都没有执行step1ChatPromptTemplate.from_template(将以下内容翻译成英文{text})step2ChatOpenAI(modelQwen/Qwen3.6-35B-A3B,temperature0,base_urlhttps://api.siliconflow.cn/v1)step3StrOutputParser()chainstep1|step2|step3# 返回 RunnableSequence尚未调用任何 API# 直到这里才真正执行outputchain.invoke({text:今天天气很好})print(output)这段代码把|的惰性特征表现得非常清楚。chain step1 | step2 | step3发生在 Python 进程内部只是创建了一个RunnableSequence对象不涉及任何 I/O 操作。直到chain.invoke(...)被调用框架才开始依次执行 prompt 模板填充、LLM API 请求和 parser 后处理三个步骤一气呵成但只有调用方自己控制时机。这种设计还有一个重要的工程收益是链对象可以被持久化、跨函数传递甚至在多个请求之间反复复用不必每次重新构建。并行处理RunnableParallel有些场景下同一份输入需要经过多条独立的分析路径而这些路径之间没有依赖关系。例如对一篇文章同时生成摘要、提取关键词、判断情感倾向三条路径可以完全并行没有理由顺序等待。RunnableParallel正是为这种场景设计的。RunnableParallel接收若干个具名子链将同一个输入分发给所有子链并发执行最终把所有结果汇总成一个字典返回字典的键就是你定义时指定的名称。这样既节省了等待时间又让结果结构一目了然。下面是一个文章三路分析的实际例子用RunnableParallel把摘要、关键词和情感判断同时发起fromlangchain_core.promptsimportChatPromptTemplatefromlangchain_core.output_parsersimportStrOutputParserfromlangchain_core.runnablesimportRunnableParallelfromlangchain_openaiimportChatOpenAIfromdotenvimportload_dotenv# 读取 .env 文件并注入环境变量之后 LangChain 会自动从环境变量中获取密钥load_dotenv()llmChatOpenAI(modelQwen/Qwen3.6-35B-A3B,temperature0.5,base_urlhttps://api.siliconflow.cn/v1)parserStrOutputParser()summary_chain(ChatPromptTemplate.from_template(用一句话总结{article})|llm|parser)keywords_chain(ChatPromptTemplate.from_template(提取3个关键词逗号分隔{article})|llm|parser)sentiment_chain(ChatPromptTemplate.from_template(判断情感倾向正面/负面/中性{article})|llm|parser)parallelRunnableParallel(summarysummary_chain,keywordskeywords_chain,sentimentsentiment_chain,)resultparallel.invoke({article:今天发布了新款手机性能大幅提升价格却比上一代更低。})# result 是一个 dict{summary: ..., keywords: ..., sentiment: ...}print(result)这里的parallel是一个顶层 chain同样可以用|与上下游组件串联。调用.invoke()时框架会将同一篇article同时发送给三条子链LLM 三次请求几乎是同时发出的最终等待最慢的那条返回后汇总为字典。如果你的业务需要更细粒度的控制RunnableParallel还支持传入匿名函数和RunnableLambda让你对同一条输入做任意维度的拆分处理。条件分支RunnableBranch真实业务往往需要根据输入的内容决定走哪条处理路径。比如客服系统需要把标记为紧急的工单路由给更快的处理流程普通工单则走常规流程。RunnableBranch提供了这种动态路由能力。RunnableBranch接收一系列(谓词, 处理链)元组外加最后一个作为默认分支的处理链。执行时它依次检查每个谓词谓词是一个接收输入并返回布尔值的函数第一个返回True的谓词对应的处理链会被调用。如果所有谓词都不匹配则使用默认分支。这与 if/elif/else 的逻辑完全一致只是写成了可以嵌入 LCEL chain 的形式。下面的客服路由例子根据输入话题中是否包含紧急字样分别走不同的 prompt 链fromdotenvimportload_dotenvfromlangchain_core.promptsimportChatPromptTemplatefromlangchain_core.output_parsersimportStrOutputParserfromlangchain_core.runnablesimportRunnableBranchfromlangchain_openaiimportChatOpenAI# 读取 .env 文件并注入环境变量之后 LangChain 会自动从环境变量中获取密钥load_dotenv()llmChatOpenAI(modelQwen/Qwen3.6-35B-A3B,temperature0,base_urlhttps://api.siliconflow.cn/v1)parserStrOutputParser()urgent_chain(ChatPromptTemplate.from_template(【紧急处理】请立即回复此问题{topic})|llm|parser)normal_chain(ChatPromptTemplate.from_template(请回复此问题{topic})|llm|parser)branchRunnableBranch((lambdax:紧急inx[topic],urgent_chain),# 谓词匹配则走紧急链normal_chain,# 默认分支)print(branch.invoke({topic:紧急服务器无法访问}))print(branch.invoke({topic:如何重置密码}))这个例子中两条.invoke()调用分别触发了不同的分支。第一个输入匹配到lambda谓词使用了带【紧急处理】前缀的 prompt。第二个输入不匹配任何条件自然落入默认的normal_chain。实际项目中你可以在分支上接入完全不同的模型、工具链甚至外部 APIRunnableBranch本身也是一个Runnable照样能用|串在更大的 chain 中。自定义组件RunnableLambdaLCEL 内置的组件覆盖了大多数常见需求但总有一些业务特定的处理逻辑例如格式转换、数据清洗、自定义聚合等需要用普通 Python 函数来实现。RunnableLambda就是连接普通函数与 LCEL 世界的桥梁。只需要把任意一个可调用对象函数、lambda、带__call__的类实例传给RunnableLambda它就变成了一个合法的Runnable可以用|自由拼接到任何 chain 中。函数的输入就是上游组件的输出函数的返回值就是传给下游组件的输入行为与其他Runnable完全一致。下面演示一个常见的后处理需求将 LLM 返回的描述词转为大写并添加标记前缀然后用|无缝拼入 chainfromlangchain_core.promptsimportChatPromptTemplatefromlangchain_core.output_parsersimportStrOutputParserfromlangchain_core.runnablesimportRunnableLambdafromlangchain_openaiimportChatOpenAI# 读取 .env 文件并注入环境变量之后 LangChain 会自动从环境变量中获取密钥load_dotenv()llmChatOpenAI(modelQwen/Qwen3.6-35B-A3B,temperature0,base_urlhttps://api.siliconflow.cn/v1)parserStrOutputParser()# 定义一个普通函数将文本转为大写并添加前缀defto_uppercase_with_prefix(text:str)-str:returnf[PROCESSED]{text.upper()}# 用 RunnableLambda 包装使其可以接入 chainpostprocessRunnableLambda(to_uppercase_with_prefix)chain(ChatPromptTemplate.from_template(用一个词描述{thing})|llm|parser|postprocess# 无缝接入自定义逻辑)resultchain.invoke({thing:太阳})print(result)# 例如[PROCESSED] BRIGHT这段代码的关键在于to_uppercase_with_prefix虽然只是一个普通函数但经过RunnableLambda包装后它获得了与ChatPromptTemplate、ChatOpenAI、StrOutputParser完全相同的接口可以通过|直接插入链的末尾。parser输出的字符串会原封不动地传入to_uppercase_with_prefix处理后的结果作为整个 chain 的最终返回值。如果需要传递额外参数可以用functools.partial先绑定好再把绑定后的函数交给RunnableLambda灵活度非常高。练习任务实现一个包含 parallel 分支的 LCEL chain编写自定义 Runnable 并集成到 chain 中尝试 stream 模式观察流式输出考核点 ✅代码验收提交包含RunnableParallel的 chain 代码输出至少 2 路并行结果自定义组件提交一个自定义RunnableLambda的 chain能正常 invoke 返回结果流式体验演示stream模式输出与invoke模式对比说明差异原理阐述口头解释|运算符在 LCEL 中的实际作用
Day 3：LCEL 深度理解
发布时间：2026/5/30 8:00:10
LCEL 是什么——声明式组合的哲学在早期版本的 LangChain 中构建一条处理链需要命令式地依次调用每个组件先调用 Prompt 生成消息再把消息传给 LLM 得到响应最后调用 Parser 解析输出。这种写法直观但随着链路变长嵌套调用和中间变量会迅速积累代码可读性和可维护性都会下降。更麻烦的是流式输出、批量处理、异步执行这些能力无法自动复用每条链都需要单独处理。LCELLangChain Expression Language用声明式的思路解决了这个问题。你只需要描述这些组件按什么顺序组合而不需要关心每一步如何传递数据。LCEL 的核心思想是组合本身就是值你用|把几个组件串在一起得到的是一个新的可执行对象流式、批量、异步等能力由框架统一提供无需重复实现。这与函数式编程中的管道pipeline概念一脉相承让链的结构一眼就能看清楚。Runnable万物皆可组合的统一接口LCEL 能够把不同类型的组件自由组合依赖的是一个统一的抽象Runnable接口。LangChain 中几乎所有的核心组件等都实现了这个接口因此它们在 LCEL 眼中是等价的积木可以任意拼接。Runnable对外暴露三种主要的调用方式分别对应不同的使用场景。invoke是最基础的同步调用接收一个输入等待整个链执行完毕后返回最终结果适合单次请求。stream则以迭代器的方式逐块返回结果LLM 每生成一个 token 就立刻推送出来用户不需要等到全部生成完才看到响应非常适合对话类应用。batch接收一个输入列表在内部并发执行多个invoke比顺序循环调用效率高得多适合需要批量处理文本的离线任务。三种方式共用同一条 chain 定义切换成本极低。下面的示例以解释一个概念的简单链为例依次演示这三种调用fromdotenvimportload_dotenvfromlangchain_core.promptsimportChatPromptTemplatefromlangchain_core.output_parsersimportStrOutputParserfromlangchain_openaiimportChatOpenAI# 读取 .env 文件并注入环境变量之后 LangChain 会自动从环境变量中获取密钥load_dotenv()promptChatPromptTemplate.from_template(用一句话解释{concept})llmChatOpenAI(modelQwen/Qwen3.6-35B-A3B,temperature0,base_urlhttps://api.siliconflow.cn/v1)parserStrOutputParser()chainprompt|llm|parser# invoke同步等待完整结果resultchain.invoke({concept:递归})print(result)# stream逐 token 打印无需等待全部完成forchunkinchain.stream({concept:递归}):print(chunk,end,flushTrue)# batch并发处理多个输入resultschain.batch([{concept:递归},{concept:闭包},{concept:协程}])print(results)这段代码先用prompt | llm | parser定义了一条简单的三步骤链。invoke是最常见的调用方式传入一个字典返回最终文本整个过程是同步阻塞的。stream则把链中的 LLM 调用切换为流式输出模式每次只返回一个 token用for循环边接收边打印用户能实时看到文字逐字出现。batch接收的是一个列表内部对列表中的每个元素发起一次独立的invoke并且这些调用是并发执行的三项概念的解释会在大致相同的时间内完成而不是按顺序逐个等待。管道运算符|的秘密|在 Python 里是位运算的按位或运算符但 LangChain 通过重载__or__和__ror__魔术方法赋予了它全新的含义。当你写step1 | step2时Python 实际上调用的是step1.__or__(step2)而Runnable的这个方法会返回一个RunnableSequence对象把两个步骤封装在一起。关键在于|只是声明不是执行。chain prompt | llm | parser这行代码执行完之后没有任何网络请求发出没有任何 token 被生成chain只是一个描述先走 prompt再走 llm最后走 parser的配置对象。真正的执行发生在你调用.invoke()、.stream()或.batch()的时候。这种惰性设计让你可以把链的构建和链的执行完全分离也让链本身可以被序列化、复用和传递。下面用一条简单的翻译链来验证这一点注意代码中注释标注的声明与执行两个阶段fromdotenvimportload_dotenvfromlangchain_core.promptsimportChatPromptTemplatefromlangchain_core.output_parsersimportStrOutputParserfromlangchain_openaiimportChatOpenAI# 读取 .env 文件并注入环境变量之后 LangChain 会自动从环境变量中获取密钥load_dotenv()# 声明三步串联的链——此刻什么都没有执行step1ChatPromptTemplate.from_template(将以下内容翻译成英文{text})step2ChatOpenAI(modelQwen/Qwen3.6-35B-A3B,temperature0,base_urlhttps://api.siliconflow.cn/v1)step3StrOutputParser()chainstep1|step2|step3# 返回 RunnableSequence尚未调用任何 API# 直到这里才真正执行outputchain.invoke({text:今天天气很好})print(output)这段代码把|的惰性特征表现得非常清楚。chain step1 | step2 | step3发生在 Python 进程内部只是创建了一个RunnableSequence对象不涉及任何 I/O 操作。直到chain.invoke(...)被调用框架才开始依次执行 prompt 模板填充、LLM API 请求和 parser 后处理三个步骤一气呵成但只有调用方自己控制时机。这种设计还有一个重要的工程收益是链对象可以被持久化、跨函数传递甚至在多个请求之间反复复用不必每次重新构建。并行处理RunnableParallel有些场景下同一份输入需要经过多条独立的分析路径而这些路径之间没有依赖关系。例如对一篇文章同时生成摘要、提取关键词、判断情感倾向三条路径可以完全并行没有理由顺序等待。RunnableParallel正是为这种场景设计的。RunnableParallel接收若干个具名子链将同一个输入分发给所有子链并发执行最终把所有结果汇总成一个字典返回字典的键就是你定义时指定的名称。这样既节省了等待时间又让结果结构一目了然。下面是一个文章三路分析的实际例子用RunnableParallel把摘要、关键词和情感判断同时发起fromlangchain_core.promptsimportChatPromptTemplatefromlangchain_core.output_parsersimportStrOutputParserfromlangchain_core.runnablesimportRunnableParallelfromlangchain_openaiimportChatOpenAIfromdotenvimportload_dotenv# 读取 .env 文件并注入环境变量之后 LangChain 会自动从环境变量中获取密钥load_dotenv()llmChatOpenAI(modelQwen/Qwen3.6-35B-A3B,temperature0.5,base_urlhttps://api.siliconflow.cn/v1)parserStrOutputParser()summary_chain(ChatPromptTemplate.from_template(用一句话总结{article})|llm|parser)keywords_chain(ChatPromptTemplate.from_template(提取3个关键词逗号分隔{article})|llm|parser)sentiment_chain(ChatPromptTemplate.from_template(判断情感倾向正面/负面/中性{article})|llm|parser)parallelRunnableParallel(summarysummary_chain,keywordskeywords_chain,sentimentsentiment_chain,)resultparallel.invoke({article:今天发布了新款手机性能大幅提升价格却比上一代更低。})# result 是一个 dict{summary: ..., keywords: ..., sentiment: ...}print(result)这里的parallel是一个顶层 chain同样可以用|与上下游组件串联。调用.invoke()时框架会将同一篇article同时发送给三条子链LLM 三次请求几乎是同时发出的最终等待最慢的那条返回后汇总为字典。如果你的业务需要更细粒度的控制RunnableParallel还支持传入匿名函数和RunnableLambda让你对同一条输入做任意维度的拆分处理。条件分支RunnableBranch真实业务往往需要根据输入的内容决定走哪条处理路径。比如客服系统需要把标记为紧急的工单路由给更快的处理流程普通工单则走常规流程。RunnableBranch提供了这种动态路由能力。RunnableBranch接收一系列(谓词, 处理链)元组外加最后一个作为默认分支的处理链。执行时它依次检查每个谓词谓词是一个接收输入并返回布尔值的函数第一个返回True的谓词对应的处理链会被调用。如果所有谓词都不匹配则使用默认分支。这与 if/elif/else 的逻辑完全一致只是写成了可以嵌入 LCEL chain 的形式。下面的客服路由例子根据输入话题中是否包含紧急字样分别走不同的 prompt 链fromdotenvimportload_dotenvfromlangchain_core.promptsimportChatPromptTemplatefromlangchain_core.output_parsersimportStrOutputParserfromlangchain_core.runnablesimportRunnableBranchfromlangchain_openaiimportChatOpenAI# 读取 .env 文件并注入环境变量之后 LangChain 会自动从环境变量中获取密钥load_dotenv()llmChatOpenAI(modelQwen/Qwen3.6-35B-A3B,temperature0,base_urlhttps://api.siliconflow.cn/v1)parserStrOutputParser()urgent_chain(ChatPromptTemplate.from_template(【紧急处理】请立即回复此问题{topic})|llm|parser)normal_chain(ChatPromptTemplate.from_template(请回复此问题{topic})|llm|parser)branchRunnableBranch((lambdax:紧急inx[topic],urgent_chain),# 谓词匹配则走紧急链normal_chain,# 默认分支)print(branch.invoke({topic:紧急服务器无法访问}))print(branch.invoke({topic:如何重置密码}))这个例子中两条.invoke()调用分别触发了不同的分支。第一个输入匹配到lambda谓词使用了带【紧急处理】前缀的 prompt。第二个输入不匹配任何条件自然落入默认的normal_chain。实际项目中你可以在分支上接入完全不同的模型、工具链甚至外部 APIRunnableBranch本身也是一个Runnable照样能用|串在更大的 chain 中。自定义组件RunnableLambdaLCEL 内置的组件覆盖了大多数常见需求但总有一些业务特定的处理逻辑例如格式转换、数据清洗、自定义聚合等需要用普通 Python 函数来实现。RunnableLambda就是连接普通函数与 LCEL 世界的桥梁。只需要把任意一个可调用对象函数、lambda、带__call__的类实例传给RunnableLambda它就变成了一个合法的Runnable可以用|自由拼接到任何 chain 中。函数的输入就是上游组件的输出函数的返回值就是传给下游组件的输入行为与其他Runnable完全一致。下面演示一个常见的后处理需求将 LLM 返回的描述词转为大写并添加标记前缀然后用|无缝拼入 chainfromlangchain_core.promptsimportChatPromptTemplatefromlangchain_core.output_parsersimportStrOutputParserfromlangchain_core.runnablesimportRunnableLambdafromlangchain_openaiimportChatOpenAI# 读取 .env 文件并注入环境变量之后 LangChain 会自动从环境变量中获取密钥load_dotenv()llmChatOpenAI(modelQwen/Qwen3.6-35B-A3B,temperature0,base_urlhttps://api.siliconflow.cn/v1)parserStrOutputParser()# 定义一个普通函数将文本转为大写并添加前缀defto_uppercase_with_prefix(text:str)-str:returnf[PROCESSED]{text.upper()}# 用 RunnableLambda 包装使其可以接入 chainpostprocessRunnableLambda(to_uppercase_with_prefix)chain(ChatPromptTemplate.from_template(用一个词描述{thing})|llm|parser|postprocess# 无缝接入自定义逻辑)resultchain.invoke({thing:太阳})print(result)# 例如[PROCESSED] BRIGHT这段代码的关键在于to_uppercase_with_prefix虽然只是一个普通函数但经过RunnableLambda包装后它获得了与ChatPromptTemplate、ChatOpenAI、StrOutputParser完全相同的接口可以通过|直接插入链的末尾。parser输出的字符串会原封不动地传入to_uppercase_with_prefix处理后的结果作为整个 chain 的最终返回值。如果需要传递额外参数可以用functools.partial先绑定好再把绑定后的函数交给RunnableLambda灵活度非常高。练习任务实现一个包含 parallel 分支的 LCEL chain编写自定义 Runnable 并集成到 chain 中尝试 stream 模式观察流式输出考核点 ✅代码验收提交包含RunnableParallel的 chain 代码输出至少 2 路并行结果自定义组件提交一个自定义RunnableLambda的 chain能正常 invoke 返回结果流式体验演示stream模式输出与invoke模式对比说明差异原理阐述口头解释|运算符在 LCEL 中的实际作用
