最近重新系统刷了一遍LangChain的LCEL和Runnable坦白说第一次接触这两个概念的时候我是有点懵的——网上教程要么只讲管道符|怎么接要么把Runnable列成一堆接口方法两者之间的关联没人讲透。后来自己动手把一条链从零搭到能跑、能流式输出、能并行分支才真正想明白Runnable是LangChain给所有组件定下的统一接口契约LCEL则是基于这个契约设计出来的声明式组合语法。这篇文章就把我梳理出来的理解路线完整写下来适合刚学完Prompt和模型调用、正准备往组合链方向走的读者也适合被各种|写法搞晕的人回来对照着看。1. Runnable到底是什么LangChain组件的统一接口1.1 从胶水代码说起先说我刚开始学LangChain时的痛点。那时候我手里的组件是散的Prompt模板要调format()模型实例要么predict()要么直接传字符串输出解析器要单独parse()。写一条简单问答链路代码长这样prompt_value prompt.format(questionquestion) raw_output model.invoke(prompt_value) # 不同模型返回类型还不一样 answer parser.parse(raw_output) print(answer)看起来也没多复杂但一旦要加缓存、加重试、加批量、加流式这些周边能力每个组件都自己实现一遍接口就越来越难统一。Runnable要解决的就是这件事给所有组件定一个共同的接口让prompt、model、parser、retriever都能被同样地调用、组合、配置。你不需要关心组件内部是什么只需要知道它实现了Runnable约定的一套方法。1.2 核心方法同步异步各三个Runnable约定的核心方法就六个同步和异步各占一半。我建议直接把这张表记下来后面所有链的操作都离不开它们方法作用典型输入典型输出invoke单次调用字符串或字典字符串或字典batch批量调用输入列表输出列表stream流式调用单个输入输出迭代器ainvoke异步单次调用字符串或字典字符串或字典abatch异步批量调用输入列表输出列表astream异步流式调用单个输入异步输出迭代器这里说的实现不是严格意义上的Java接口继承LangChain用的是鸭子类型加协议约定。所以自定义组件的最低要求是让对象具备invoke等方法实际项目里更常见的是继承Runnable基类或者直接用后面会讲的RunnableLambda把普通函数包装成Runnable。1.3 万物皆Runnable组合结果还是RunnablePromptTemplate是RunnableChatModel是RunnableStrOutputParser是RunnableRetriever是Runnable甚至Tool也是Runnable。它们内部实现天差地别但对外暴露的接口一致。这带来一个特别重要的推论由多个Runnable组合出来的链本身也是一个Runnable。这个特性怎么理解就像USB接口一样不管里面是硬盘、手机还是键盘插上同一个口就能用只不过这里的插口是invoke/batch/stream这一组方法。更关键的是你拼出来的整条链也带着这个USB口——它可以继续被别的链当作一个组件拼进去。理解了这一点再看LCEL就会顺很多LCEL其实就是在说怎么用|把这些Runnable像积木一样拼起来拼出来的东西还能继续拼。2. LCEL的管道语法为什么一条链能自动继承所有能力2.1|运算符的底层语义a | b在LangChain里并不是什么魔法它等价于RunnableSequence(firsta, lastb)也等价于a.pipe(b)。多个|连写就是连续套娃最终生成一条串行执行的链。这里有个关键点a的输出必须能作为b的输入。LangChain不会帮你做隐式类型转换所以最常见的报错就是上一段吐出字符串下一段却要字典或者反过来。2.2 最小可运行的示例先看一条最基础的链用户给一个概念模型用一句话解释它。from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser llm get_llm() # 接入你自己的模型实例 prompt ChatPromptTemplate.from_template(用一句话解释{concept}。) parser StrOutputParser() chain prompt | llm | parser result chain.invoke({concept: LCEL}) print(result)后面所有示例我都用llm占位不绑定具体供应商。你平时用的那家服务无论是云端API还是本地部署都会提供一个LangChain封装好的模型类实例化后填到这里就行。这条链的执行流程很直白invoke({concept: LCEL})传给prompt模板模板格式化出完整提示词模型拿到提示词返回原始输出parser把模型输出解析成纯字符串。每一步的输出恰好是下一步的输入所以能串起来。2.3 为什么推荐LCEL而不是手写Python函数很多人会问我直接用Python函数串流程不好吗为什么非要搞一套语法我的体会是LCEL最大的价值不是省几行代码而是能力继承。可读性prompt | llm | parser一眼看懂数据流向手写函数要一行行读才知道在干嘛。接口自动齐全你拼出的链立即拥有invoke/batch/stream和异步版本不用自己实现任何东西。中间态可观测插一个RunnableLambda(lambda x: print(x) or x)进去就能看某一步的输出。非业务能力可挂载重试、回退、并发限制这些统统用配置加上去不用改链的结构。手写Python函数最大的问题在于每换一个需求比如从单次调用改成批量就要重写一遍控制逻辑。LCEL把这些控制逻辑全部标准化了你只需要关心业务本身。这一点到我后来做并行分支和流式输出的时候感受特别深。3. 四类核心Runnable串行、并行、函数封装与透传的选型3.1 RunnableSequence所有|链的底层只要你用了|底层就是RunnableSequence。直接new一个出来在需要动态拼接链的场景里很常见两种写法完全等价from langchain_core.runnables import RunnableSequence # 方式一管道符 chain prompt | llm | parser # 方式二显式构造 chain RunnableSequence(firstprompt, middle[llm], lastparser)实际开发中我基本都用|显式构造只会在链的结构本身需要由配置动态生成时用。比如一个系统里有十几条链它们的中间步骤根据用户选项变化这时候用RunnableSequence去拼就比写死一串|灵活得多。3.2 RunnableParallel并行分支与结果合并RunnableParallel用来做互不依赖的并行分支。它的输出是一个字典key就是你在构造时指定的分支名。举个常见的场景同一个主题同时生成中文和英文介绍。from langchain_core.runnables import RunnableParallel zh_chain prompt_zh | llm | parser en_chain prompt_en | llm | parser parallel RunnableParallel(zhzh_chain, enen_chain) result parallel.invoke({topic: LCEL}) print(result[zh]) print(result[en])注意两个使用前提第一分支之间不能有数据依赖否则别用Parallel老老实实串起来第二所有分支共享同一个输入数据需要在各自内部把需要的字段取出来。RunnableParallel在默认实现里会尽量并发调度这几个分支但你不必太纠结底层线程模型把它理解成结构上分叉、结果上合并就够了。3.3 RunnableLambda与RunnablePassthrough普通函数和透传RunnableLambda是把普通函数变成Runnable的万能包装器项目中几乎必用。它有两点要注意函数接收的是整份输入数据如果输入是字典函数拿到的就是字典本身。from langchain_core.runnables import RunnableLambda def clean_text(text: str) - str: return text.strip() clean RunnableLambda(clean_text) chain prompt | llm | parser | cleanRunnablePassthrough则简单得多原样透传输入不修改任何东西。它最常用的玩法是配合.assign()在透传的同时追加新字段from langchain_core.runnables import RunnablePassthrough chain RunnablePassthrough.assign( upperlambda x: x[text].upper() )这个模式特别适合既要保留原始输入又要往里塞中间计算结果的场景。比如后面实战里我要把用户原问题一路带到最终步骤就是用RunnablePassthrough透传的。顺带一提条件分支可以看RunnableBranch它像一个if/elif/else分发表但日常用得没有前几个频繁。4. invoke/batch/stream与异步版本六个方法的用法矩阵4.1 同步三件套单次、批量、流式同步场景里invoke是最容易上手的batch和stream则各有讲究。看代码# 单次调用 result chain.invoke({question: 什么是LCEL?}) # 批量调用 results chain.batch([ {question: 什么是LCEL?}, {question: 什么是Runnable?}, ]) # 流式调用 for chunk in chain.stream({question: 什么是LCEL?}): print(chunk, end)batch非常适合离线处理一批问题默认会尽量并发执行实际并发度可以通过config{max_concurrency: 2}这类参数限制。stream则需要注意一点LCEL链会尝试把流式能力逐段传递下去但如果链中间某个组件不支持流式输出那个组件会退化成一次性计算完整结果后续组件再继续流式。所以不要以为写了stream就一定能逐token输出得看你用的模型封装和自定义组件支不支持。4.2 异步三件套Web服务里的刚需异步版本在Web服务场景几乎是必须的否则一个慢模型请求会卡住整个事件循环。FastAPI接口里处理一个对话请求正确姿势长这样import asyncio async def handle_request(question: str): async for chunk in chain.astream({question: question}): # 推给前端 yield chunk asyncio.run(handle_request(什么是LCEL?))ainvoke和abatch同理区别只是把流式换成完整结果和批量结果。我见过不少同学学的时候只练同步三件套上了生产才发现接口全被堵死又回来补异步的课。建议从一开始就把同步和异步六件套一起写进代码里反正只是加一个a前缀的事。4.3 运行配置重试、回退与可观测性Runnable真正值钱的是那些非业务能力可以直接挂载。三个最常用的配置方法我列一下# 给链起名字方便日志和链路追踪 chain.with_config({run_name: qa_chain}) # 自动重试模型偶发超时很有效 chain.with_retry( retry_if_exception_type(TimeoutError,), stop_after_attempt3, wait_exponential_jitterTrue, ) # 兜底回退主模型挂了切备用 chain.with_fallbacks([backup_chain])尤其是with_fallbacks生产环境必备。主模型临时不可用的时候链自动切到备用通道用户几乎无感知。这就是我前面说的能力继承在实际项目里的价值——这些逻辑要是手写每个链条都要复制一遍还容易出bug。5. 实战并行分支问答链的搭建与三个典型报错排查5.1 需求拆解现在把前面的知识串起来做一个稍微完整的场景用户输入一个问题我希望生成一份高质量的最终答复。拆成两条并行分支一条生成初步回答一条提取问题关键词最后用一个汇总模型把初步回答和关键词整合成最终答案。这样设计的好处是初步回答负责内容正确性关键词提取负责信息结构化汇总阶段再把两者融合比单次调用的效果更稳。5.2 完整代码与数据流转from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser from langchain_core.runnables import RunnableParallel, RunnablePassthrough llm get_llm() # 换成你自己的模型实例 prompt_draft ChatPromptTemplate.from_template( 请简洁回答下面的问题\n{question} ) prompt_keys ChatPromptTemplate.from_template( 请从下面的问题中提取3个核心关键词用逗号分隔\n{question} ) prompt_final ChatPromptTemplate.from_template( 你有一段初步回答和一组关键词请把它们整合成一段结构完整、可读性强的最终答复。\n 初步回答{draft}\n 关键词{keys}\n 原始问题{question}\n 最终答复 ) draft_chain prompt_draft | llm | StrOutputParser() keys_chain prompt_keys | llm | StrOutputParser() parallel_branch RunnableParallel(draftdraft_chain, keyskeys_chain) chain ( {question: RunnablePassthrough()} | parallel_branch | prompt_final | llm | StrOutputParser() ) question LCEL和Runnable是什么关系 for chunk in chain.stream(question): print(chunk, end)这里的关键是开头的{question: RunnablePassthrough()}。因为用户输入是字符串而后续组件需要字典所以在链首先做一个字符串转字典的映射。如果输入本身已经是字典也可以用operator.itemgetter(question)这类方式取字段效果一样。整条链的数据流转看这张表阶段输入输出{question: RunnablePassthrough()}字符串问题{question: ...}parallel_branch{question: ...}{question: ..., draft: ..., keys: ...}prompt_final上述字典提示词文本llm提示词文本模型原始输出StrOutputParser()模型原始输出最终字符串我某次运行时流式输出拼接出来的示意结果大致是LCEL是LangChain基于Runnable接口设计的声明式组合语法而Runnable是所有组件共用的接口契约。可以把Runnable理解为管道中的标准接头LCEL则是用这个接头快速拼接管道的方式。第一段答案给出了两者关系的基本判断关键词部分则点出了接口、组合、语法这些核心概念两者结合后得到的最终答复就显得结构清晰、主次分明。5.3 三个典型报错与排查链路这条链麻雀虽小五脏俱全我实际调试过程中遇到过的报错基本都能归类成下面三类排查思路也一并写出来。报错一unsupported operand type(s) for |原因几乎都是管道符某一侧不是Runnable对象。比如写着llm | 字符串或者llm | some_function字符串和普通函数都不能直接放|右边。排查方法是把每个片段单独输出看类型自定义函数先用RunnableLambda包一层再接入链。报错二Missing some input keys: {question}这类报错原因就是链的输入数据和PromptTemplate需要的变量对不上。最常见的是chain.invoke(一个问题字符串)但prompt模板里写的是{question}LangChain不知道字符串该填到哪个变量。排查思路是先看报错里缺哪些key然后在链首加{question: RunnablePassthrough()}做类型转换或者检查RunnableParallel的分支字段名是否和模板变量完全一致。字段名差一个字符就会触发这个错误我在这上面栽过不止一次。报错三stream不流式一次性吐出一大块原因基本是链里夹了一个不支持流式的组件。比如某些自定义后端模型封装、或者一个会聚合结果的RunnableLambda。排查办法是逐段测试先只跑prompt | llm看有没有流式输出再往后加组件定位到哪一段开始退化。还有一个隐藏坑如果最终链里包含batch之类的聚合操作stream也会被压成一次性输出。调试方面两个工具很实用chain.get_graph().print_ascii()可以把链的结构画出来一眼看出串联顺序有没有问题在链中间插RunnableLambda(lambda x: print(x) or x)可以随时观察某一步的数据形态确认字典的key和值对不对。先拿小数据跑通正确性再上流式验证是我比较推荐的调试顺序。6. 学习路线从LCEL语法到Runnable心智模型的顺序建议6.1 我建议的顺序先跑通再拆解最后读协议如果让我重新学一遍我会给自己定这么个顺序。第一步先把prompt | llm | parser这种最基础的三段链跑通感受组合结果本身也是Runnable这件事。第二步刻意练习数据结构的转换字符串怎么变成字典、字典的key怎么在链中间增加、RunnableParallel合并出的字典长什么样。这一步最枯燥但最值钱因为大部分报错都出在数据形态不匹配上。第三步去看Runnable类的方法定义重点理解invoke/batch/stream之间的约定不用逐行读源码但至少要明白每个方法默认行为是什么。这个顺序帮我避开了死记API却拼不出链的弯路。6.2 版本与依赖的坑LangChain迭代很快0.1到0.2再到0.3之间的API一直有变动有些早期教程里的from langchain.llms import ...写法现在已经废弃import路径经常变。我的建议是锁定版本再学习比如pip install langchain某个稳定版本同时保证langchain-core的版本和主包对齐。版本不一致很容易出现照着教程写明明没问题却报ImportError的情况这时候先看看是不是版本差异导致的。固定版本还有一个好处网上提问时别人能根据版本号快速定位问题而不是来回猜。6.3 最后一点个人体会把LCEL和Runnable放在一起看我的最终体会是Runnable是契约LCEL是语法真正难的是流程设计——怎么把业务拆成能被组合的单元。拆得好链会非常干净拆得不好就会写出塞满RunnableLambda的缝合怪。判断标准其实就一条每个单元的输入和输出类型是否清晰稳定。抓住这条主线报错、调试、扩展都会变简单很多。后面无论是要做带记忆的链还是带工具调用的Agent核心都是在Runnable这一层做组合。先把这一层打牢再往上层走会轻松很多。写这篇文章的时候我又翻了一遍自己早期的笔记很多当时觉得玄学的报错现在看全都是输入输出类型没对齐的问题。希望这篇梳理能帮你少走一段我走过的弯路。