★ FEATURED ARTICLE LangChain 升到 v1.0 之后最让老用户不适应的一个变化就是Chain 这个单词几乎从 API 里消失了。以前我们嘴边挂着 LLMChain、ConversationChain、SequentialChain现在去看官方文档推荐的做法都是围绕 Runnable 和 LCEL 展开。这篇文章不绕弯子直接把这套新范式拆开讲Runnable 到底是什么、LCEL 怎么把链路像拼水管一样拼起来、以及我实际迁项目时踩过的坑和总结出的排查方法。正在折腾 v1.0 的朋友或者之前被各种 Chain 类搞晕的新手都可以拿这篇当一份可操作的参考指南。1. 为什么 v1.0 要把 Chain 重构为 Runnable LCEL1.1 旧 Chain 的三大痛点在 Runnable 这套抽象普及之前LangChain 的链路基本是靠各种 Chain 类堆出来的。早期版本里LLMChain、SimpleSequentialChain、ConversationChain、RetrievalQAChain 这些类各自维护一套调用方式有的用run()有的用apply()有的用generate()。我刚接触的时候最头疼的就是每换一个 Chain都要重新记它的方法签名和返回类型文档里还经常看到“deprecated请迁移到 LCEL”这样的提示。看起来只是命名和 API 不统一实际上暴露了更深的问题。第一链与链之间不能方便地组合。SimpleSequentialChain 只能做一维串联一旦出现“并行分支”“条件分支”这种结构就得绕路去写复杂回调甚至退回去手写 LLM 调用逻辑。第二流式输出支持很差。很多旧 Chain 内部用llm.predict()这类一次性方法把整个 answer 全部生成完才返回用户体验大打折扣尤其是想要打字机效果的时候根本没法做。第三可观测性不足。链路一旦长了中间某一步出错定位问题的成本很高。这些痛点本质上是同一个问题链路没有被抽象成一个统一的、可组合的组件协议。Chain 类只是在 LLM 调用外面包了一层壳并没有形成一套标准的数据流描述方式。1.2 Runnable 协议一次抽象处处统一v1.0 把这件事收拢到了 Runnable 这个基类上。核心思路非常简单不管你是提示词模板、大模型、输出解析器、向量检索器还是一个自定义的普通 Python 函数只要实现了 Runnable 接口就能统一使用invoke()、stream()、batch()这三个方法以及对应的异步版本ainvoke()、astream()、abatch()。可以把它想象成电源插座标准。以前每个设备都有自己的专用插头现在所有 Runnable 都是同一个规格的插头不管后面接的是什么插上去就能通电。这个抽象让 LangChain 的组件生态在 v1.0 里第一次有了统一的交互协议。更重要的是Runnable 接口还内置了一批实用能力with_config()可以给节点调整参数、with_retry()可以给链路增加自动重试、with_fallbacks()可以在主链路失败时切到备用链路、bind()可以给模型绑定额外参数。这些功能都是接口层面的通用方法不需要每个组件单独实现。我迁移旧代码时最直观的感受是以前想要“失败了自动换一个模型再试一次”还得自己去 try/except现在直接.with_fallbacks([backup_model])就完事。1.3 LCEL 的精髓声明式管道LCEL 的全称是 LangChain Expression Language它并不是一套新的运行时而是把 Runnable 接口包装成语义清晰的声明式语法。核心操作符只有一个|管道符。a | b的含义就是把 a 的输出作为 b 的输入最终返回一个新的 RunnableSequence。很多人第一次见到prompt | model | parser时会觉得这只是一个语法糖。但从工程角度看它带来的价值比“少写几行代码”大得多LCEL 天然支持并行分支、条件路由、流式传播而且链路本身仍然是一个 Runnable可以继续被别的链路当作普通节点组合。这种“任意链路可嵌套、可复用”的特性才是 LCEL 能被称作范式的原因。实际开发中我特别喜欢的一点是LCEL 链路的可读性非常高。一个复杂的 RAG 流程用旧 Chain 可能要写几十行初始化和调用代码用 LCEL 表达出来基本是“从输入格式化成检索上下文再拼 prompt再进模型再解析输出”这样能直接读下来的结构。我迁完一个项目后回头看链路长什么样数据在每个节点之间怎么流动一眼就能看清不再需要去翻中间变量。2. LCEL 组合核心五个组件一次讲透2.1 RunnableSequence管道符背后的默认容器管道符|左右两边的对象最终都会被包装进一个 RunnableSequence。RunnableSequence 是一个有序执行的列表前一个节点的输出会作为后一个节点的输入。最简单的例子from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser from langchain_openai import ChatOpenAI model ChatOpenAI(modelgpt-4o-mini, temperature0) prompt ChatPromptTemplate.from_messages([ (system, 你是一个擅长解释概念的助手。), (human, 请用通俗的语言解释{topic}) ]) chain prompt | model | StrOutputParser()调用chain.invoke({topic: LCEL})时实际发生的事是prompt 节点拿到 dict渲染成消息列表model 节点拿到消息列表调用模型返回一个 AIMessageStrOutputParser 把 AIMessage 里的字符串 content 提取出来作为最终结果返回。值得注意的是RunnableSequence 也是懒加载的。构建链条时不会真的执行任何东西只有调用invoke()等方法时才触发计算。这意味着你可以把整条链当作变量传来传去在运行时再决定是否拼接后续节点这给动态构建复杂流程留下了很大空间。2.2 RunnableParallel并行分支的利器在很多实际场景中一个问题的回答可能依赖多个独立子任务的结果。比如做产品分析时需要同时生成“优点总结”和“缺点总结”这两个子任务互相没有依赖完全可以并行执行。RunnableParallel 干的就是这件事。from langchain_core.runnables import RunnableParallel summary_chain prompt | model | StrOutputParser() keyword_chain keyword_prompt | model | StrOutputParser() parallel_chain RunnableParallel( summarysummary_chain, keywordskeyword_chain ) result parallel_chain.invoke({topic: LangChain v1.0}) # result 是一个 dict # {summary: ..., keywords: ...}这里有一个很容易忽略的细节dict 字面量本身就是 RunnableParallel 的简写。也就是说可以直接在 LCEL 管道里写{summary: summary_chain, keywords: keyword_chain}来替代显式创建 RunnableParallel。两种写法等价但 dict 形式与管道组合阅读起来更流畅complex_chain { summary: summary_chain, keywords: keyword_chain, } | combine_prompt | model | StrOutputParser()并行执行的代价是并发控制。这个我在后面“踩坑”部分再展开这里先记住一个原则分支之间没有依赖关系时再用 RunnableParallel不要为了“结构好看”强行把一个串行流程拆成并行否则反而会增加调试难度。2.3 RunnablePassthrough 与 assign透传和字段改造RunnablePassthrough 名字很直白就是“原样通过”。它的典型用途有两个。第一个用途是占位透传。比如上面的 RunnableParallel 写法里如果希望输入的 question 原封不动传给后续节点可以这样写from langchain_core.runnables import RunnablePassthrough chain ( { question: RunnablePassthrough(), context: retriever, } | prompt | model | StrOutputParser() )这里question分支会把自己接收到的输入原样保存到结果 dict 里context分支会把同一份输入丢给 retriever让 retriever 返回检索文档列表。这种写法在做 RAG 时非常常见。第二个用途是.assign()方法。它会在保留原 dict 所有字段的基础之上新增或覆盖字段chain ( RunnablePassthrough.assign(lengthlambda x: len(x[question])) ) result chain.invoke({question: hello}) # 结果是 {question: hello, length: 5}需要特别注意.assign()返回的是新 dict不会原地修改原始输入。如果你在 RunnableLambda 里写x[new_field] ...然后 return x虽然也能跑但这个行为破坏了“函数式变换”的直觉在并行或重复调用时容易产生副作用。我的建议是尽量使用.assign()或显式构造新 dict别去复用输入 dict 做原地修改。2.4 RunnableLambda 与 RunnableBranch把任意函数变成节点系统设计里永远有定制需求RunnableLambda 就是用来把普通 Python 函数包装成 Runnable 的桥接器。任何“输入是一个值输出是一个值”的函数都可以变成一个链路节点。from langchain_core.runnables import RunnableLambda def format_docs(docs): return \n\n.join(doc.page_content for doc in docs) chain ({ context: retriever, question: RunnablePassthrough(), } | RunnableLambda(lambda inputs: { question: inputs[question], context: format_docs(inputs[context]), }) | prompt | model | StrOutputParser())如果函数内部有重逻辑建议显式给 RunnableLambda 命名比如RunnableLambda(format_docs).with_config({run_name: format_context})生成的追踪日志会好读很多。RunnableBranch 则是条件路由。它接收一组(条件函数, 对应链路)元组按顺序判断条件命中哪个就执行哪条链路最后还允许提供一个默认链路兜底。from langchain_core.runnables import RunnableBranch short_chain short_prompt | model long_chain long_prompt | model branch RunnableBranch( (lambda x: len(x[text]) 20, short_chain), (lambda x: len(x[text]) 20, long_chain), fallback_chain, )这里有个新手很容易搞混的点条件函数返回的是布尔值而不是链路对象。整个元组的结构必须是(condition, runnable)RunnableBranch 内部会自己调用 condition根据返回值决定走哪条。条件函数的输入就是上游传下来的整个数据流对象所以一般会写成接收 dict 并进行字段判断。2.5 组合规则速查为了减少记忆负担我把这个阶段最常用的组合规则整理成一张表你想实现的效果推荐组合方式串行处理a 的输出给 ba多个子任务并行合并成 dict{key1: a, key2: b}或 RunnableParallel保留原字段同时新增处理结果RunnablePassthrough.assign(...) 或{...}把普通函数放进链路RunnableLambda(func)按条件选择不同链路RunnableBranch((condition, chain), ..., default_chain)给下游统一补充固定参数.bind(**kwargs)比如model.bind(stop[\n])3. 从 LLMChain 到 LCEL一份能直接照抄的迁移手册3.1 迁移前的旧代码长什么样以前写一个“给文档生成摘要”的链路老代码大概是这样的from langchain.chains import LLMChain from langchain.prompts import PromptTemplate from langchain.llms import OpenAI llm OpenAI() prompt PromptTemplate( input_variables[document], template请给下面的文档写一段摘要\n{document}, ) chain LLMChain(llmllm, promptprompt) result chain.run({document: doc_text})看起来不算复杂但一旦链路变成“先把文档切块再逐块摘要再汇总成总摘要”就得靠多个 LLMChain 拼成一个 SimpleSequentialChain或者用 SequentialChain 手动指定 input_variables 和 output_variables。代码很快膨胀到难以维护。更麻烦的是旧 Chain 的run()方法只关心第一个输出变量如果你的 prompt 有多个输出变量还需要用return_only_outputs之类的参数去调整行为。3.2 新老 API 对照表我迁项目时做了一张对照表基本能覆盖绝大多数常见用法旧 Chain / 旧方法新 Runnable / 新方法LLMChain(llmllm, promptprompt)prompt | modelchain.run(inputs)chain.invoke(inputs)chain.apply(input_list)chain.batch(input_list)chain.predict(**kwargs)chain.invoke({key: value})SimpleSequentialChainchain1 | chain2SequentialChain管道传 dict或用 RunnableParallel 组合字段ConversationChain显式维护 history或用 RunnableWithMessageHistorychain.generate(inputs)chain.stream(...)/chain.batch(...)verboseTruechain.get_graph().print_ascii() callbacks这里我想单独强调 SequentialChain 的迁移方式。旧 SequentialChain 的核心是每步可以指定 input_variables 和 output_variables中间结果会被保留并传给后续步骤。在 LCEL 里这个模式对应的就是“管道中传递 dict每个节点只负责读取自己需要的字段并返回新的 dict”。step1 step1_prompt | model step2 step2_prompt | model sequential_chain ( RunnablePassthrough() | step1 | step2 )如果 step1 的输出字段名和 step2 的输入字段名不一致通过一个 RunnableLambda 做字段映射就行比旧 SequentialChain 的“变量自动传递”更显式、更好控制。3.3 推荐迁移顺序盲目把旧代码一口气改成 LCEL容易在语法和语义上踩坑。我建议按下面这个顺序来每一步都验证之后再进下一步先梳理当前链路的所有步骤写清楚每一步的输入结构和输出结构。这一步不需要看代码先在纸上或文档里画出数据流哪些是串行、哪些是并行、哪些是条件分支。用prompt | model | parser的方式重建单个完整链路。把原来的 LLMChain 拆成 prompt 定义、模型实例、输出解析三步先跑通一个最小可运行链路确保输入输出和旧逻辑一致。处理并行和分支。如果原来的链路里有“多个独立子任务合并到同一个结果”的逻辑引入 RunnableParallel如果有条件判断引入 RunnableBranch。用chain.get_graph().print_ascii()检查全链路结构。这个方法可以直接在终端打印出链路拓扑图节点之间的连接关系一目了然比读代码强太多了。最后做回归测试。不要一次性删掉旧代码用同样的输入同时跑新旧两套链路对比输出和耗时确认行为一致后再清理旧实现。4. 实战搭一条支持流式输出的 RAG 问答链路4.1 链路设计某个模拟的内部文档问答项目需求是用户输入一个问题系统从本地知识库检索相关文档片段把片段整理成上下文交给大模型生成答案。同时前端希望答案能像打字机一样逐字输出。数据流拆解下来是这样的原始 question 字符串去检索器生成 document 列表把 document 列表格式化成一段上下文文本再连同 question 一起填充进 prompt交给模型最后解析输出。4.2 完整实现代码先准备一个简单的向量存储和检索器。这里假设已经用某种 embedding 模型完成了文档入库from langchain_core.documents import Document from langchain_core.vectorstores import InMemoryVectorStore from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser from langchain_core.runnables import RunnablePassthrough, RunnableLambda from langchain_openai import ChatOpenAI # 假设 docs 是已经切好的文档片段列表 vector_store InMemoryVectorStore.from_documents(docs, embeddingembedding_model) retriever vector_store.as_retriever(search_kwargs{k: 3}) prompt ChatPromptTemplate.from_messages([ (system, 你是一个严谨的文档问答助手。请只根据提供的上下文回答不要编造。), (human, 问题{question}\n\n上下文\n{context}), ]) def format_docs(docs): return \n\n.join(f[来源]\n{doc.page_content} for doc in docs) rag_chain ( { context: retriever, question: RunnablePassthrough(), } | RunnableLambda(lambda inputs: { question: inputs[question], context: format_docs(inputs[context]), }) | prompt | model | StrOutputParser() ) answer rag_chain.invoke(LangChain v1.0 的 Runnable 是什么)这段代码里的重点是第二段匿名函数它接收一个 dict其中 context 字段是检索器返回的 Document 列表question 字段是原始问题字符串。这个函数负责把 Document 列表转成文本并重新拼成一个只包含 question 和 context 两个字段的 dict供 prompt 使用。4.3 增强流式、重试与兜底这个链路现在虽然能工作但还缺生产环境需要的几个能力。首先是流式输出。因为 LCEL 的流式机制会自动沿着链路传播模型层如果支持 stream那么整条链的.stream()就能逐 chunk 输出async for chunk in rag_chain.astream({question: Runnable 是什么}): print(chunk, end, flushTrue)然后是失败兜底。我给生产链路加了两层保护一层是.with_retry()针对超时异常自动重试另一层是.with_fallbacks()新增一个备用模型链路当主链路完全不可用时自动切换fallback_model ChatOpenAI(modelgpt-4o-mini, temperature0, max_retries0) fallback_chain prompt | fallback_model | StrOutputParser() robust_chain rag_chain.with_retry( stop_after_attempt2, retry_if_exception_type(TimeoutError, ConnectionError), ).with_fallbacks([fallback_chain])把这些增强能力组合在一起后对外暴露给前端的依然只是一个 Runnable。前端不需要知道背后有检索、有格式化、有重试、有兜底只需要调用robust_chain.astream(...)。这正是 Runnable 抽象最大的工程价值复杂的内部编排被封裝成一个标准接口消费者只关心输入和输出。5. 高频踩坑与排查技巧5.1 流式输出静悄悄没了很多人在自己组装 LCEL 链路后发现.stream()不生效所有内容一次性返回。最常见的原因就是链路中某个位置使用了 RunnableLambda 对模型输出做了一次“全部接收再返回”的处理。模型输出的流式是按 token 推进的。StrOutputParser 保留了这个流式特性但一个普通的 RunnableLambda 默认会等上游全部产出后才调用函数函数返回单个值后再交给下游流式在这里被截断。解决办法有几种如果后处理只是简单的字符串提取优先用官方 parser如果不是简单提取可以考虑自定义 Runnable 实现 stream 方法最退一步干脆在流式路径上避免使用 RunnableLambda把后处理改成在下游手动async for拿到完整内容后再统一处理。我实测过的经验是链路里如果还有检索、格式化这些步骤模型输出本身并不受影响真正破坏流式的是模型后面那个 RunnableLambda。排查时可以先在后处理节点之前调用.stream()如果这里输出正常问题基本就锁定在后处理节点。5.2 在 RunnableLambda 里调用其他 Runnable这是一个非常隐蔽的坑。有些人写了这样的代码def func(inputs): result another_chain.invoke(inputs[question]) return {answer: result} chain RunnableLambda(func)从功能上它能跑但它彻底破坏了 LCEL 的层级结构。another_chain 相当于被藏在一个黑盒函数里它内部的节点不会出现在整个链路图中callbacks、retry、fallback 这些能力也没法传播到内部链路。更严重的是如果 another_chain 内部又调用了别的 Runnable很容易触发递归层数超限问题。正确的姿势是把 another_chain 直接放在管道中让它作为独立节点参与组合chain ( RunnablePassthrough() | another_chain | RunnableLambda(lambda result: {answer: result}) )我的习惯是RunnableLambda 里只做纯数据变换所有涉及 LLM 调用、检索、外部 IO 的逻辑都封装成正常 Runnable 放在组合表达式里不要让函数偷偷去调用别的链路。这样链路图才能完整反映真实执行流程。5.3 同步异步混用导致事件循环被卡死在 fastapi 或异步脚本里很容易写出同步阻塞的 RunnableLambda。比如函数内部用time.sleep(1)模拟延迟或者调用一个同步的 requests 请求。这个同步函数被丢进事件循环线程后会阻塞整个循环其他并发任务全部卡住。我踩过这个坑后的处理方式是如果 RunnableLambda 被用在异步链路中函数内部用await asyncio.to_thread包装重逻辑或者直接给 RunnableLambda 同时提供同步和异步两个实现。LangChain 对 RunnableLambda 支持func和afunc两个参数存在异步版本时会优先使用异步版本。另一个小细节是如果链路在 Jupyter 或脚本里用.ainvoke()跑但前面的 embedding 或重试又依赖同步客户端也容易出现“看起来没响应”的现象。最稳妥的做法是统一定位排查整个链路里所有可能做 IO 的节点确认它们都是同一门派的。混合使用同步和异步客户端不是不行但需要你在 RunnableLambda 边界上做显式线程切换别指望框架替你解决。5.4 并行分支并发过高触发限流RunnableParallel 虽然很好用但它默认对并发分支数量不做限制。如果你的并行分支里都是 LLM 调用一次invoke()可能同时发出 N 个请求很容易触发服务端的限流。解决方案也很直接在调用.batch()时通过 config 控制并发results chain.batch( [{question: 问题一}, {question: 问题二}], config{max_concurrency: 5} )如果并行分支本身不想限制但分支内部需要对某个共享资源做限流可以在相应节点头上用.with_config({max_concurrency: 2})来控制。这个配置不影响链路结构只影响运行时并发所以很适合在压测阶段动态调参。5.5 调试三板斧LCEL 链路在报错时最尴尬的往往是“不知道数据在哪一步变形了”。我调试时基本用三板斧。第一板斧是chain.get_graph().print_ascii()。这条命令能打印链路结构快速确认节点顺序和分支是否正确尤其是类型不匹配的报错看一眼图基本就知道是哪两个节点衔接出了问题。第二板斧是挂回调。给链路传config{callbacks: [handler]}在 on_chain_start、on_chain_end 里打印每个节点的输入输出。注意传输的数据可能很大比如检索器返回的 document 列表所以我自己通常只打字段名和前几百个字符。第三板斧是在关键节点插一个“间谍节点”def peek(value): print(fpeek: {value}) return value chain ... | RunnableLambda(peek) | ...这个办法虽然简单但非常有效。我在生产代码里一般不会保留这种 spy调试完就删。最后说一个我自己的体会。LCEL 看着简单但它真正的价值不是省几行代码而是强制你把应用拆成一个个有明确输入、有明确输出的节点。以前我在 RunnableLambda 里塞过各种外部调用、状态管理结果一到流式和并发就翻车。现在我的规矩是RunnableLambda 只做纯数据变换所有 IO 都封装成独立 Runnable 再组合。养成了这个习惯之后链路跑通率和调试效率都提升了一大截我觉得这也是 v1.0 新范式最值得吸收的经验。
阅读完成 · 觉得有帮助？
