LangChain LCEL 核心模式与组合一、概念速查LCEL 核心组件组件类型作用类比Runnable基类所有可执行单元的抽象接口函数签名RunnableSequence组合器链式调用前一个输出是后一个输入管道 RunnableParallel组合器多个 Runnable 并行执行合并输出Promise.allRunnablePassthrough工具透传输入或调用函数后合并到输出中间件RunnableLambda适配器把任意 Python 函数包装为 Runnable函数包装RunnableBranch路由根据条件选择不同执行路径if/elseRunnableMap工具对输入做字段映射和转换map各组件核心方法# Runnable 接口三方法invoke(input,config)# 同步执行ainvoke(input,config)# 异步执行batch(inputs,config)# 批量执行stream(input,config)# 流式输出astream(input,config)# 异步流式代码示例从一个 Prompt 到完整 Chainfromlangchain_core.runnablesimportRunnablePassthroughfromlangchain_core.output_parsersimportStrOutputParserfromlangchain_core.promptsimportChatPromptTemplatefromlangchain_openaiimportChatOpenAI promptChatPromptTemplate.from_template(用一句话解释{concept})modelChatOpenAI(modelgpt-4o-mini)parserStrOutputParser()chainprompt|model|parserprint(chain.invoke({concept:LCEL}))# 输出类似LCEL 是 LangChain 的声明式组合语法用 | 运算符将组件串联成可执行的 Runnable 链。并行执行与字段映射fromlangchain_core.runnablesimportRunnableParalleldefextract_keywords(x:str)-dict:return{keywords:x[:20]}chainRunnableParallel(explanationprompt|model|parser,metadataRunnablePassthrough()|RunnableLambda(extract_keywords))resultchain.invoke({concept:RunnableParallel})print(result[explanation])# LLM 输出print(result[keywords])# RunnableParallel二、底层原理Runnable 协议架构RunnableSequenceRunnableParallelRunnableBranchRunnableLambdainvoke(input)Runnable 基类类型分发逐个执行子节点并行执行全部节点条件匹配后执行直接调用 Python 函数合并输出返回最终结果LCEL 的核心抽象是一个统一的Runnable协议。任何对象只要实现了invoke/ainvoke/batch/stream四个方法就可以通过|运算符组合。运算符重载的实质是构造一个RunnableSequence内部维护了一个有序的子节点列表执行时按序传递输出。这种设计解决了传统回调式框架的两个根本问题一是链中每个环节都可以单独测试任何 Runnable 都可以独立 invoke二是组合关系与执行逻辑解耦运行时可以通过中间件拦截任意节点的输入输出。RunnableSequence 的执行模型classRunnableSequence:def__init__(self,*steps):self.stepsstepsdefinvoke(self,input,configNone):forstepinself.steps:inputstep.invoke(input,config)returninput每一步接收上一步的输出作为输入。对于ChatPromptTemplate输入是 dict输出是ChatPromptValue。对于ChatOpenAI输入是ChatPromptValue输出是ChatGeneration。对于StrOutputParser输入是ChatGeneration输出是str。LCEL 自动处理这些类型转换开发者不需要手动适配。RunnableParallel 的 fork-join 模式输入RunnableParallel分支 A分支 B分支 C合并Dict 输出RunnableParallel本质上是一个 fork-join 原语。输入被广播到所有分支每个分支独立执行结果以字段名聚合为 dict。这在需要同时做多件事的场景中极其有用例如同时生成文本解释和提取元数据。流式处理的实现机制forchunkinchain.stream({concept:流式处理}):print(chunk,end,flushTrue)流式的底层依赖生成器协议。每个 Runnable 节点可选择实现stream方法返回迭代器。RunnableSequence在流式模式下会逐级传递生成器最后一个节点产出流式 token。StrOutputParser是流式友好的——它逐个 yield 字符而不是等全部生成再返回。三、架构设计原则1. 组合优于继承所有行为通过 Runnable 组合实现而不是通过子类化扩展。新增功能意味着接入新的 Runnable而不是修改现有类的继承树。2. 每个 Runnable 可独立测试assertisinstance(prompt.invoke({concept:test}),ChatPromptValue)assertisinstance(model.invoke(prompt.invoke({concept:test})),AIMessage)这条原则直接来源于 Runnable 协议的统一性。如果某段 LCEL 链出错了你可以从中间截取任意节点独立验证而不是从头跑到尾 debug。3. 类型安全优先于运行时容错LCEL 在__or__运算符中不做类型检查——类型不匹配会在 invoke 时报错而不是在构造时静默吞掉。这鼓励开发者在组合阶段就发现接口不匹配而不是留到生产运行时爆。4. Config 透传与中间件能力每个 Runnable 的invoke方法都接受一个config字典框架用它透传回调、标签、元数据。这意味着你可以在不修改任何 Runnable 实现的前提下通过 config 注入日志、追踪、限流等横切关注点。5. 延迟执行优于即时求值LCEL 链在构造时只是建立计算图不会执行任何实际计算。prompt | model | parser只是构造了一个RunnableSequence对象直到调用invoke才真正触发执行。这让链可以被缓存、序列化、动态修改后再执行。