LangChain v1.0 的 Chain 新范式核心就俩词Runnable 接口和 LCEL。我在实际项目里切换过来之后最大感受是“之前写的链像是在叠积木现在像是在接水管”——每个组件都有统一的接口数据从一个组件流到下一个组件哪里断了、哪里慢了一眼能定位。这篇指南不是概念复述而是把 Runnable 接口和 LCEL 从原理到实操、从调用方式到排查问题完整过一遍。适合正在从旧版本升级的同学也适合直接从新范式入手的新手建议边看边把代码跑一遍。1. 新旧 Chain 的碰撞v1.0 到底革了什么命1.1 旧 Chain 设计的三处硬伤在 LangChain v1.0 之前的很长一段时间里写一条链的常规做法是找一个现成的 Chain 子类继承它重写内部方法。比如最早接触的 LLMChain、RetrievalQAChain、ConversationChain每个类都有自己的 run、apply、generate 之类的接口。这种设计的第一个大问题是组合靠继承。想做一个“先检索、再总结、然后翻译”的流程就得硬生生造一个新的 Chain 子类把三个逻辑揉进一个类里。类多了以后整个项目的链类膨胀得特别快改一个公共逻辑所有子类都可能受影响。第二个问题是调用方式割裂。有的链用 run 调用有的链用 predict有的链用 generate同步和异步方法散落在不同的基类上。项目如果同时存在几条不同来源的链调用层就得写一堆适配代码把各自的方法统一包装一遍。更麻烦的是流式输出基本没被认真设计过很多链要么不支持逐 token 输出要么只能靠外部轮询模拟体验很差。第三个问题是状态管理不透明。旧版链条里经常带着 memory、callback、verbose 这些全局旁路这些状态在链内部怎么流转对外是不透明的。一旦链在中间某一步报错或者输出不符合预期排查非常痛苦因为每一步的输入输出都散落在各个类内部没有统一观察点。所以 v1.0 这次“范式革命”本质上不是加了一两个新类而是把链从“类继承的产物”改成了“协议组合的产物”。理解这一点后面所有用法都能顺下来。1.2 Runnable一个接口取代一整套类体系新范式的底座是一个叫 Runnable 的接口。注意是接口不是基类。这意味着你想让任何东西变成链条的一部分不需要继承它只需要让自己的类实现几个约定好的方法。可以把它理解成 USB-C 口充电器、显示器、硬盘、键盘都做成同一个物理接口插上就能用不需要为每种设备单独设计一种接口。Runnable 就是这个物理接口任何组件——模型、提示词模板、解析器、检索器、普通函数——只要实现了 Runnable 协议就能被放进同一条链里和别的组件自由拼接。具体来说一个 Runnable 对象要提供的核心能力是接收一个输入返回一个输出。输入和输出可以是字符串也可以是字典只要是协议支持的合法类型。在 v1.0 里更常见的是字典因为要在链条中间传递多个变量只有字典才装得下。接口本身不关心你的组件内部干了什么它只保证一件事我可以用统一的方式调用你。这个“统一”带来的好处是连锁的因为所有组件都是 Runnable所以组合工具本身也是 Runnable比如用 LCEL 构建出的整条链本质上还是一个 Runnable它可以被当成单个组件嵌到更大的链里。链套链、嵌套组合天然成立不需要额外的层级抽象。1.3 核心方法速览一张表看清调用全貌Runnable 接口约定了一组方法分别是同步调用、批量调用、流式调用以及它们对应的异步版本方法作用典型场景注意事项invoke(input)接收一个输入返回完整输出单次请求问答、工具调用最常用新手先掌握这个batch(inputs)接收一组输入批量返回一组输出离线批量处理、压测底层有并发调度注意并发上限stream(input)接收一个输入返回迭代器逐段产出打字机效果、长文本生成需要模型和中间组件都支持流式ainvoke(input)invoke 的异步版本在异步服务里调用配合 asyncio不要和同步调用混用abatch / astream对应上面两个的异步版本异步批量、异步流式注意所在运行循环的限制这些方法的存在意味着同一套链可以无缝切换调用姿势。比如用户交互时用 stream后台任务里用 batchWeb 接口里用 ainvoke切换成本几乎为零因为链的定义部分完全不用动。我在实际项目里最喜欢的一点是invoke 和 stream 可以同时存在于同一条链上不会互相破坏状态。以前用旧版链我时不时要担心 run 之后内部是不是残留了什么计数、缓存、临时变量现在只要遵循协议组件的生命周期非常干净。这种稳定性直接降低了维护成本。注意异步方法名统一加 a 前缀这是反直觉但规律性很强的设计。写异步代码时优先用 ainvoke不要在异步循环里硬调同步 invoke否则会阻塞事件循环。2. LCEL用管道符把链条“写”出来2.1 管道符的本质是数据流合约LCELLangChain Expression LanguageLangChain 表达式语言是 v1.0 里构建链的推荐方式核心语法就是管道符 |。管道符的语义非常直观A | B 表示把 A 的输出作为 B 的输入。写起来和 Unix 命令行管道一样左边产数据右边收数据。比如 prompt | model 就是一个最简链prompt 格式化出提示词model 根据提示词生成回复。但这里有一个容易忽略的点| 不是 Python 原生操作符的简单复用它是 Runnable 协议提供的一个合约。只有当左右两边都是 Runnable 对象时管道符才有意义两边拼出来的结果本身也还是一个 Runnable可以继续往右接。因此 LCEL 的本质是把“数据如何流转”用一段线性描述写出来而不是用一堆命令式代码去手动组装。这种声明式写法的最大价值是链的结构和意图完全对上。我拿到一段 LCEL 代码不用去读每个组件的内部实现光看管道顺序就知道数据从哪来到哪去中间经过了哪些转换。对协作和维护来说这比旧版那种“看类名猜行为、读代码追状态”舒服太多。2.2 三个高频组合子Passthrough、Lambda、ParallelLCEL 里除了直接拼组件还有几个高频组合子应该先掌握RunnablePassthrough透传组件。输入是什么输出就是什么。唯一的作用是把数据原样往后送通常配合 assign 方法使用用来在字典里加字段而不丢掉已有字段。RunnableLambda函数适配器。把一个普通 Python 函数包成 Runnable。它是接入“自定义逻辑”的万能口比如格式化字符串、清洗数据、调用外部工具。RunnableParallel并行容器。接收一个字典字典的每个 value 都是一个 Runnable所有 value 并行执行最后把各自的输出合并成一个更大的字典。这三个组合子解决的是连接问题。没有它们LCEL 只能做“模型对模型”的拼接有了它们普通函数、外部 API、分支逻辑都能被织进链里而且不会破坏 Runnable 这个统一协议。我举个例子说明 RunnablePassthrough 的必要性。假设一条链最开始接收的输入是一个字典里面有 question 和 history 两个字段。模型只需要 question但后续的解析器还需要 history 做输出格式化。这时候如果直接把 question 传给模型history 就丢了。用 RunnablePassthrough.assign(question...)可以在保留 history 的前提下覆盖或新增 question 字段后面的组件还能继续拿到完整字典。这种细节写复杂链的时候特别实用不然变量传着传着就没了。2.3 手写一个完整 LCEL 链检索问答实例光讲语法不够我直接给一个可以跑通的检索增强问答链把 Runnable、LCEL、组合子全串起来from langchain_core.runnables import RunnablePassthrough, RunnableParallel, RunnableLambda from langchain_core.output_parsers import StrOutputParser from langchain_core.prompts import ChatPromptTemplate # 假设 retriever 已经构造好它应该是一个实现了 Runnable 协议的检索器 # retriever ... # 用你自己的检索器实例替换 # model ... # 用任意一个实现了 Runnable 的聊天模型实例替换 def format_docs(docs): return \n\n.join([d.page_content for d in docs]) prompt ChatPromptTemplate.from_template( 根据下面的资料回答问题。\n资料{context}\n问题{question}\n答案 ) # 第一步把输入字典拆成两个分支一个去检索一个原样保留 question setup RunnableParallel( contextretriever | RunnableLambda(format_docs), questionRunnablePassthrough() ) # 第二步拼接整条链 chain ( setup | prompt | model | StrOutputParser() ) result chain.invoke({question: LangChain v1.0 的新范式是什么}) print(result)这段代码的执行流程是invoke 传入一个字典setup 里的 context 分支去调用 retriever拿到文档后经过 format_docs 变成文本question 分支直接透传。两个分支的结果合并成新字典{context: ..., question: ...} 交给 prompt。prompt 渲染出完整提示词model 生成回复最后 StrOutputParser 把模型回复里的内容字符串提取出来。注意这里 retriever 必须也是一个 Runnable否则没法直接放在管道里。v1.0 的检索器基本都实现了这个接口如果你自己的检索器是普通函数用 RunnableLambda 包一层再放进来就行。LCEL 的妙处在这个例子里体现得很充分并行分支、函数转换、模板渲染、模型调用、输出解析全部用线性管道描述出来每一段都能独立替换不用改其他部分。真要说维护体验的提升这是我迁移后感受最深的一点。3. Runnable 实操四种调用姿势与组合技巧3.1 invoke、stream、batch、ainvoke 各自的使用边界在真实项目里我发现很多人只记得 invoke其他方法基本不用其实每个调用姿势都有明确的使用场景。invoke 最稳适合交互简单、一次性拿到完整结果的场景。比如在命令行工具里跑一条链、在测试脚本里验证链是否正常。stream 适合需要“打字机效果”的场景比如聊天机器人、内容生成的渐进展示。batch 适合离线批处理比如给 100 条用户评论逐个做情感分析它在底层做了并发调度比循环里挨个 invoke 快得多。ainvoke 适合把链嵌到异步服务端接口里用 async def 的接口直接 await 链的执行。一个链可以同时支持这四种用法这是 Runnable 协议带来的红利。同样的链对象# 同步单次调用 result1 chain.invoke({question: 什么是 LCEL}) # 流式迭代调用 for chunk in chain.stream({question: 讲一下 Runnable 接口}): print(chunk, end, flushTrue) # 批量调用 results chain.batch([ {question: 什么是 LCEL}, {question: Runnable 是什么}, ]) # 异步单次调用 import asyncio async def main(): result4 await chain.ainvoke({question: 异步调用怎么做}) return result4这里有一个容易踩的坑stream 能不能真正流出 token取决于链上每一个组件的实现。如果模型本身支持流式但中间的解析器不是生成器型的结果可能会被缓冲成一个大块输出。排查方法后面专门讲。batch 的并发度也不是无限大的它底层会开线程池或进程池默认线程数受运行时和配置影响。大批量任务时先测一下最优 batch 大小开太大反而会因为上下文切换变慢。3.2 分支、并行、回退、重试组合子的正确打开方式链并不永远是线性管道真实场景里经常要“如果怎么怎么样就走哪条路”。RunnableBranch 就是用来做条件路由的它的参数是若干 (条件, 分支) 对加上一个默认分支条件和分支都是 Runnable。from langchain_core.runnables import RunnableBranch classify RunnableBranch( (lambda x: len(x[text]) 100, long_handler), # long_handler 是某个 Runnable (lambda x: len(x[text]) 20, medium_handler), # medium_handler 同理 short_handler, # 默认分支 )这里的每个 handler 都是一个 Runnablelambda 作为条件判断。RunnableBranch 本身也是 Runnable所以可以继续在管道右边接东西。回退和重试也是生产环境的刚需。with_fallbacks 可以在主链抛异常时自动切换到备选链比如模型 A 超时了自动换模型 Bwith_retry 则是在失败时按指数退避重试。两者可以叠加先重试再回退。我通常给外部模型调用配置成重试 2 次间隔从 1 秒起失败后回退到本地规则链这样至少不会让用户拿到空响应。safe_chain ( primary_chain .with_retry(stop_after_attempt3, wait_exponential_jitterTrue) .with_fallbacks([fallback_chain]) )这些组合子全部返回 Runnable意味着你可以把它们继续嵌在 LCEL 管道里。对于长期运行的机器人服务来说没有这些兜底线上迟早会暴雷。我自己就在一个内部工具上吃过亏没有配 fallback上游模型服务抖动一次整个请求链条直接报错用户端看到的就是一个红色的异常页面。后来加了回退链体验立刻稳住了。3.3 调试链的两个实用技巧调试 LCEL 链最朴素也最有效的方法是“插桩”在怀疑的环节后面接一个 RunnableLambda把输入或输出打出来。def debug_step(x): print( DEBUG ) print(x) return x # 沿用前面定义的 setup 和 prompt / model / StrOutputParser chain ( setup | RunnableLambda(debug_step) | prompt | model | StrOutputParser() )注意 debug_step 必须返回原来的 x否则链的数据流就被破坏了。生产环境里去掉这个节点即可。另一个技巧是控制单步超时。网络请求类组件可以用 .with_timeout(seconds) 包一层避免整条链因为某个上游接口卡死而长时间无响应。这对于对外服务尤其重要省得用户端一直转圈。提示在链的中间层加调试节点是定位问题最快的方式比看日志猜变量直接得多。等确认无误后再移除调试节点。4. 常见问题与排查技巧实录4.1 stream 只出整段结果原因逐个排除这是我在社区里看到最多的问题也是我自己早期踩过的大坑。明明用了 stream屏幕上还是一下子打印一整段完全没有打字机效果。排查顺序很重要。先看模型层。不是所有模型服务商都对流式接口开放或者你的客户端连接设置了全局缓冲区。如果直接调用模型本身就不流式返回那么上游链做得再对也没用。再看解析层StrOutputParser 这类解析器一般会对流式结果做重组如果解析器实现里调用了完整内容再解析流到那里会被堵住。最后看链的中间件有没有哪个环节把流式迭代器整个收集成了列表。排查时我通常从模型开始逐层向上检查而不是一上来就怀疑链的写法。另一个容易被忽略的点是有些自定义 RunnableLambda 函数内部用了类似 .join(chunks) 的逻辑这等于把流提前消费完了后面自然只能拿到整段内容。所以写转换函数时尽量保持逐块处理不要主动聚合除非你确定需要。4.2 Missing key 报错其实是字典协议没对齐新范式几乎所有链的输入输出都是字典。报错信息里最常出现的就是某个 key 缺失比如 KeyError: context。绝大多数情况下不是模型不行而是上一个组件输出的字典里根本没有这个键。判断方法很简单在报错之前插一个调试节点打印输入看键名到底长什么样。很多问题都是因为用字符串索引取错了字段或者前一个组件里把字段命名成了 question_text后续组件却读取 question。修正方法有两种一是统一命名在源头上保持键名一致二是用 RunnableLambda 写一个字典转换函数把旧键名映射到新键名。def rename_fields(data): return {question: data[question_text]}4.3 fallback 写了却不触发这个坑比较隐蔽。很多人给主链配了 with_fallbacks测试时手动抛异常发现能切过去但线上模型超时却没有任何反馈。问题通常出在异常类型上fallback 默认只捕获配置时指定的异常类型如果你的模型调用抛出的异常不在列表里它就当作“无事发生”。另一个原因是 fallback 绑错了对象。with_fallbacks 是加在某个具体 Runnable 上的如果那个 Runnable 本身一直没有被真正执行到——比如链在它之前的环节就挂掉了——fallback 自然不参与。所以调试 fallback 时先在异常出现的位置附近手动触发确认它真的在这个组件的调用链上。判断标准很简单在 fallback 链里加一个调试打印如果没触发说明异常类型或绑定点有问题。4.4 性能别在循环里反复造链最后说一个性能相关的常见错误。有些项目会在 for 循环里反复调用一个内部函数每次调用都先建链再 chain.invoke(...)。链的构建本身是有成本的虽然单次看起来很快但在循环里放大后就会拖慢整体速度。正确做法是在模块加载时或者函数外先构建一次链对象循环里只调用。批量任务优先用 batch 而不是 for invoke两者差一个数量级都很正常。如果需要异步批量就在异步代码里用 abatch不要在异步函数里写 for await ainvoke。我自己做过一个简单的对比测试同样是处理 50 条输入用 batch 比循环 invoke 快了三倍多代码还更短。注意批量调用时如果每个输入都用相同的链链对象是线程安全的可以放心复用。这不是微妙的状态共享而是 Runnable 协议设计要求的一部分。5. 从旧写法迁移到新范式的分步经验5.1 新旧写法对照一个典型链的完整换血为了看清迁移前后的差异我拿最典型的 LLMChain 举例。旧版写法大致长这样简化from langchain.chains import LLMChain llm_chain LLMChain( llmmodel, promptprompt, ) result llm_chain.run(question什么是 LCEL)新版用 LCEL 重构则是chain ( {question: RunnablePassthrough()} | prompt | model | StrOutputParser() ) result chain.invoke({question: 什么是 LCEL})表面的变化是类少了、管道短了更深层的变化是调用方式从专用 run 方法收敛到了统一的 Runnable 协议。注意新版里开头那个字典字面量LCEL 会把它的每个 value 自动解释成并行分支所以 {question: RunnablePassthrough()} 其实是一个 RunnableParallel 的简写。以后想给这个链加检索、加分支、加回退不需要改链的整体结构只需要在管道里插入对应组件。5.2 迁移最小步骤清单四步走策略如果你有一个旧项目要迁我的经验是不要一次性重写按下面的步骤走先把旧链的输入输出格式固定下来明确输入字典有哪些键、输出应该是什么类型。用 LCEL 把主流程搭出来先不接复杂分支与回退跑通一个最小闭环。再逐个接入原本依赖的记忆、检索、工具调用等外围能力每接一个就做一次回归验证。最后统一替换调用点把 run 改成 invoke把需要流式输出的场景切到 stream把内部循环改造成 batch。每一步都保留旧代码做对照测试直到新链覆盖了所有旧链的场景再删掉旧代码。迁移过程中最怕的就是新链表面上跑通了但某个旁路分支没覆盖到。我在给一个内部工具做迁移的时候就漏了一个只在特定错误路径里触发的旧链结果灰度两天后才发现线上有一小部分请求还在走旧逻辑。从那以后我迁移任何链条都会先列一个场景清单一条条确认。5.3 我在迁移中踩过的三个坑第一个坑是返回值类型变化。旧链往往直接返回字符串新链更倾向于返回字典。这个变化在日志和解析逻辑里特别容易埋雷尤其是下游已经写了 isinstance(result, str) 判断的代码迁完直接全部走不到预期分支。建议所有新链强制规定输出格式并在出口加一个最终转换节点。第二个坑是旧版记忆类不能直接平替。以前用的 Memory 类本质上维护会话状态状态是全局旁路式的。LCEL 推荐的做法是把历史记录显式地作为输入键传给提示词用 RunnablePassthrough.assign 组合进去。这样虽然多写了几个字段但状态的流动路径变得完全可见排查对话问题时轻松很多。第三个坑是自定义 Chain 子类的 run 方法“遗毒”。如果你的项目里有过自定义 Chain迁移后很可能残留一些对 run 的引用或者子类实现了 run 但没实现 invoke。Runnable 协议要求的是 invoke 一族方法不实现它们链在中间层就会直接报 NotImplemented 之类的错误。所以自定义组件时别继承旧 Chain 了直接实现 Runnable 接口或者用 RunnableLambda 包函数就行。我个人在实际操作中的体会是LangChain v1.0 这套新范式真正厉害的地方不是语法好看而是把链从“黑盒类”变成了“白盒数据流”。你可以在任何一步插入调试、加分支、做兜底而不需要理解整个类体系的内部实现这让我改代码时终于敢下手了。最后再分享一个小技巧如果你是从旧版本迁过来的建议先在测试环境里把所有旧链跑一遍记录输出格式再用 LCEL 逐条还原别急着在生产上试探。全套折腾下来你会明显感觉到代码行数在缩水流式输出变稳定排查问题也变得有迹可循。对于新项目我的建议是直接按新范式来写别走回头路。