很多接触 LLM 应用开发的人应该都有过这种体验看官方文档学 LangChain 时前面几页还能跟上一旦读到 LCEL 和 Runnable 这一块就开始犯晕。网上中文资料倒是不少但大部分要么是照着官方文档念一遍要么只讲某一两个 API 的用法很少有人能讲清楚 LCEL 这套设计到底在解决什么问题为什么官方要死推这套写法以及它背后的 Runnable 协议长什么样。我一开始也是这么磕磕绊绊过来的。最开始写 Agent 和 Chain 的时候用的还是旧版的LLMChain、SimpleSequentialChain那一套。那时候写链子就像拼积木每一步都要手动调用chain.run()或者chain.apply()数据流要靠自己维护想加个并行分支、想给某一步加个重试逻辑都得自己做一堆包装。后来换到 LCEL还是抱着怀疑态度试了一段时间——结果越用越觉得这玩意儿是真的把 LangChain 的框架逻辑理顺了。这篇东西我不会照着文档抄一遍重点讲三件事LCEL 和 Runnable 解决了什么问题、它们内部是怎么工作的、怎么在真实项目里把 Runnable 玩得像样。顺带把我踩过的坑全部摊开来能帮你省不少时间。1. 先搞清楚 LCEL 和 Runnable 到底是什么关系1.1 从一段传统写法到管道写法的对比很多初学者最大的困惑在于LCEL 是一种语言吗Runnable 是一个类吗为什么网上有人把两者混着说我的理解是Runnable 是一套统一接口协议LCEL 是基于这套协议设计的声明式组合语法。Runnable 定义了一个组件应该长什么样、能被怎么调用LCEL 则定义了这些组件之间怎么用管道操作符|串起来。两者相辅相成谁也离不开谁。为了讲清楚先看一个最朴素的例子。假设我要做一个根据关键词生成一句营销文案的小工具传统写法旧版风格from langchain.llms import OpenAI from langchain.prompts import PromptTemplate from langchain.chains import LLMChain llm OpenAI(modelgpt-3.5-turbo-instruct) prompt PromptTemplate.from_template(围绕{keyword}写一句吸引人的营销文案。) chain LLMChain(llmllm, promptprompt) keyword 智能手表 text chain.run(keywordkeyword) print(text)LCEL 写法from langchain_core.runnables import RunnablePassthrough from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate prompt ChatPromptTemplate.from_template(围绕{keyword}写一句吸引人的营销文案。) llm ChatOpenAI(modelgpt-3.5-turbo) chain prompt | llm keyword 智能手表 print(chain.invoke({keyword: keyword}))功能上看两个都能跑通。但 LCEL 版本里的prompt | llm本质上是在构建一个执行图而不是调用链。这个差别很关键——它意味着你可以把prompt | llm当成一个整体再|上别的东西也可以在中间任意位置插入并行、条件分支、流式输出甚至直接把整条链子丢给另一个 Runnable 去嵌套。1.2 Runnable 的接口协议到底是什么所谓协议我的通俗解释是LangChain 规定凡是实现 Runnable 接口的对象都必须提供几个固定的调用方法并且遵守同样的输入输出约定。具体来说就是下面这几个方法invoke(input)单次调用输入一个字典或字符串返回一个结果batch(inputs)批量调用输入一组数据返回一组结果stream(input)流式调用输入单个数据返回一个迭代器逐个产出结果ainvoke/abatch/astream对应的异步版本bind(**kwargs)绑定参数生成一个新的 Runnablewith_config()/with_fallbacks()/with_retry()生成带配置的副本。只要一个对象实现了这套接口它在 LangChain 的生态里就是一个一等公民。所有组件——包括 PromptTemplate、模型、输出解析器、检索器、Agent、以及你自己用RunnableLambda包装的函数——在这个协议下都是同一种东西。这种设计带来的最大好处是抽象一致性。你不需要关心某个组件内部是什么实现只需要知道它接受什么输入、产出什么输出、能不能流式、能不能并行。就像 USB 接口一样不管里面是硬盘还是手机只要插口一致都可以连电脑。2. 为什么说管道符|是 LCEL 的灵魂2.1 从数据流的角度理解组合prompt | llm这行代码在视觉上传递了一个核心信息数据从左边流到右边左边的输出会作为右边的输入。LangChain 内部处理这一步时会自动把左边的 Runnable 执行后的结果重新包装成右边 Runnable 能接受的输入格式。初学的时候最容易忽略的是管道符并不要求两边输入输出格式完全一致LangChain 会尝试兼容。比如 PromptTemplate 的输出是PromptValue而语言模型接收的也是PromptValue但如果你llm | output_parser语言模型的输出是一个字符串输出解析器也会按约定好的格式解析。这种格式转换约定被封装在每一个 Runnable 的实现里。再进一步你可以把输入格式兼容理解成有一层隐式的适配器在背后工作。比如chain prompt | llm | StrOutputParser()StrOutputParser做的事情就是把语言模型的输出AIMessage对象变成纯字符串。没有这一步的话你会拿到一个自带 metadata 的消息对象——很多新手在这里困惑半天。2.2 为什么声明式组合比命令式调用更适合复杂流程命令式写法的核心问题是每加一个步骤就要多写一堆胶水代码而且步骤之间的数据传递全靠手工维护变量。比如上面那个营销文案的例子如果我再加一个环节——判断这段文案是否合格不合格就重写命令式写法就会变成def generate_and_check(keyword): text chain.run(keywordkeyword) if len(text) 20: text chain.run(keywordkeyword 请写长一点) return text这还只是最简单的循环。如果再加并行、加分支、加条件路由手工维护会很快变成一场灾难。而 LCEL 的写法可以很优雅地组合from langchain_core.runnables import RunnableBranch rewrite_prompt ChatPromptTemplate.from_template(现在的文案是{text}请把它扩充到20字以上。) check_chain rewrite_prompt | llm | StrOutputParser() branch_chain RunnableBranch( (lambda x: len(x[text]) 20, check_chain), lambda x: x[text] ) full_chain prompt | llm | StrOutputParser() | branch_chain这种组合子风格的好处是每一步只做一件事数据流是显式的分支条件是声明式的而且后面加需求时不需要改前面的代码。2.3 管道符背后的实际执行逻辑很多人会好奇|到底在 Python 里做了什么答案并不神秘——LangChain 重写了__or__魔法方法准确说是Runnable.__or__调用这个魔法方法时返回一个新的RunnableSequence对象。我观察它的核心逻辑大致是把管道符左边的 Runnable 作为 sequence 的第一个元素管道符右边的 Runnable 追加到 sequence 里真正执行invoke()时挨个执行每个 Runnable把前一个的输出作为后一个的输入如果某个 Runnable 支持流式输出整个 chain 会尝试构建一条端到端的流式管道前提是中间的每个组件都支持。这意味着prompt | llm | parser在内存里其实等价于一个RunnableSequence对象只不过你写代码的时候用管道符表达出来了而已。理解了这层你就明白为什么链子可以嵌套——RunnableSequence本身也是一个 Runnable它可以继续被|到别的地方。3. Runnable 的变体们Parallel、Passthrough、Lambda、Branch3.1 RunnableParallel 到底解决什么问题RunnableParallel 是我在生产项目里用得最频繁的组件之一。GPT 系列模型本身是纯文本输入输出但现实业务里经常需要把一件事拆成多件事并行处理然后再汇总。举个例子假设我在做一个商品评论分析的后端服务输入一条评论需要同时做三件事情感判断、关键词提取、是否值得回复的分类。命令式写法是顺序调用三个链子耗时大概是三次调用之和。用RunnableParallel可以让三个链子同时执行from langchain_core.runnables import RunnableParallel sentiment_chain sentiment_prompt | llm | StrOutputParser() keyword_chain keyword_prompt | llm | StrOutputParser() reply_chain reply_prompt | llm | StrOutputParser() parallel_chain RunnableParallel( sentimentsentiment_chain, keywordskeyword_chain, should_replyreply_chain ) result parallel_chain.invoke({comment: 这手机续航很强但拍照一般售后态度差。})结果会是一个字典含有sentiment、keywords、should_reply三个字段。我在实际项目里会把这种RunnableParallel嵌到更大的流程里比如后面再接一个汇总提示词把三个结果合并成最终的结构化报告。3.2 RunnablePassthrough 被低估的两个用法RunnablePassthrough字面意思就是直接传过去。它不修改数据原样转发输入。听起来很没用但实际有两个非常常见的用途。用法一作为并行分支中的数据旁路。还是拿上面的评论分析举例。如果我希望最终的汇总链能同时看到原始评论和前面三个分析结果就需要用RunnablePassthrough把原始输入原样带过来from langchain_core.runnables import RunnablePassthrough final_prompt ChatPromptTemplate.from_template( 评论内容{comment} 情感分析{sentiment} 关键词{keywords} 是否建议回复{should_reply} 请基于以上信息生成一份回复建议。 ) final_chain ( RunnableParallel( commentRunnablePassthrough(), sentimentsentiment_chain, keywordskeyword_chain, should_replyreply_chain ) | final_prompt | llm | StrOutputParser() )这里如果不用RunnablePassthroughcomment字段就会丢失后面的提示词模板根本拿不到原始评论。用法二配合assign做字段追加。RunnablePassthrough.assign()可以给当前数据字典追加新字段同时保留原字段这在做数据清洗时非常方便chain ( RunnablePassthrough.assign( keyword_numlambda x: len(x[keywords].split(,)) ) | some_prompt | llm )3.3 RunnableLambda把普通函数变成一等公民这是万物皆 Runnable的关键桥梁。LangChain 的组件再丰富也总有自己的逻辑需要塞进去。比如我在做文档问答时经常需要对检索到的文档按相关度排序、截断、去重——这些用 LangChain 原生组件没有现成方案最简单的方式就是写一个普通函数然后用RunnableLambda包装from langchain_core.runnables import RunnableLambda def deduplicate_docs(docs): seen set() unique_docs [] for doc in docs: if doc.page_content not in seen: seen.add(doc.page_content) unique_docs.append(doc) return unique_docs dedup RunnableLambda(deduplicate_docs)一个容易被忽略的细节被RunnableLambda包装的函数它的输入输出类型会被 LangChain 用来推断整个链子的类型签名。如果你在函数里使用了类型注解IDE 补全和错误提示体验会好很多。更重要的是如果类型是Iterator或者AsyncIteratorLangChain 还会尝试做流式适配。还有一点bind方法在这个组件里同样适用。比如我想让一个普通函数固定在某个参数上运行可以这样写def format_result(content, style): return f{style}风格{content} formatted RunnableLambda(format_result).bind(style正式)3.4 RunnableBranch条件路由的正确打开方式RunnableBranch 可以被理解成 LCEL 版的if-elif-else写法上很直接from langchain_core.runnables import RunnableBranch def is_short(text): return len(text) 20 branch RunnableBranch( (is_short, short_chain), (lambda text: len(text) 50, medium_chain), long_chain # 兜底 )需要注意两个使用细节。一是条件函数的入参是上一个 Runnable 传过来的数据所以条件函数的签名要和上游输出匹配。二是多个条件的顺序很重要——RunnableBranch会按顺序依次判断一旦命中就短路执行不会再走后面的分支。我当时在做客服工单自动分类时就用RunnableBranch根据工单标题长度和关键词把工单分成简单咨询需要人工介入紧急事件三条路径分别对接不同的处理链。这种需求如果用原始if-else写在业务代码里链条一旦长了就很难维护但封装成 Runnable 之后整条链路本身也可以被嵌套进更大的流程。3.5 其他值得留意的 Runnable 类除了上面这些还有几个我平时也会用到的成员RunnableMap其实就是RunnableParallel的旧版别名现在一般用RunnableParallelRunnablePick从字典输入里挑出部分字段传给下一个 RunnableRunnableAssign给数据并行追加新字段等价于RunnablePassthrough.assign()的显式版RunnableWithMessageHistory给链子自动附带历史消息管理做聊天机器人时很有用RunnableRetry自动带重试机制的包装器。每个包装器解决的都是某一类具体问题最重要的是理解它们在协议下的行为遇到实际需求时能想到这东西有现成的 Runnable 可以处理而不是又回去写命令式流程。4. 在一个问答系统里把 LCEL 真正用起来4.1 从需求到 Runnable 组合的拆解过程理论讲了一堆不如看一个完整的实战案例。假设我要做一个基于本地文档回答问题的内部知识库问答系统最核心的流程是用户输入问题系统判断问题是否需要检索知识库简单问题直接回答需要的话从向量数据库检索 top-K 文档把问题 检索到的文档拼进提示词调用模型生成回答输出结果并附带引用的文档信息。这个流程如果用传统代码实现写一个函数依次执行也可以但两个痛点会很快出现一是流程扩展性差比如以后想加个问题重写步骤、想对检索结果做重排二是调试和可观测性低每一步的输入输出得自己打日志看。用 LCEL 重构的话我先把每一步拆成独立 Runnable再组合起来from langchain_core.runnables import RunnableParallel, RunnablePassthrough, RunnableLambda from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate # 步骤1路由判断——是否需要检索 def need_retrieval(question: str) - bool: # 这里可以接一个轻量分类器也可以基于关键词规则 keywords [是什么, 怎么, 如何, 哪个, 多少] return any(k in question for k in keywords) def route_question(data): question data[question] if need_retrieval(question): return retrieve return direct retriever vectorstore.as_retriever(search_kwargs{k: 3}) def retrieve_docs(question: str): return retriever.invoke(question) # 步骤2组两套提示词 direct_prompt ChatPromptTemplate.from_template(直接回答{question}) retrieve_prompt ChatPromptTemplate.from_template( 基于以下资料回答问题 {context} 问题{question} ) # 步骤3组装并行检索 def format_docs(docs): return \n\n.join(doc.page_content for doc in docs) retrieve_chain ( RunnableParallel( questionRunnablePassthrough(), context({question: lambda x: x} | RunnableLambda(retrieve_docs) | RunnableLambda(format_docs)) ) | retrieve_prompt | llm | StrOutputParser() ) direct_chain direct_prompt | llm | StrOutputParser() # 步骤4用RunnableBranch做路由 qa_chain RunnableBranch( (lambda data: route_question(data) retrieve, retrieve_chain), direct_chain ) result qa_chain.invoke({question: 我们的报销流程是怎么样的})这段代码里最核心的是RunnableBranch 的条件判断放在了最外层接收的数据是{question: ...}整个字典。所以你需要在route_question里从 data 中取 question。这一点新手很容易写错直接写lambda x: need_retrieval(x)结果发现传进来的不是字符串而是字典导致判断逻辑永远出错。4.2 把可观测性注入 Runnable 管道真实生产环境里能跑通只是第一步。更折磨人的是——模型偶发输出格式不对、检索有时候返回空、某次调用超时……问题出在链条的哪一环LangChain 提供了一个侵入性很低的手段在 Runnable 上挂config里的callbacks。我常用的做法是给关键节点加一个简单的日志回调把进入某步、耗时多久、输入输出是什么打印到控制台或写入日志系统from langchain_core.callbacks import BaseCallbackHandler class MyLogCallback(BaseCallbackHandler): def on_chain_start(self, serialized, inputs, **kwargs): print(f 启动节点: {serialized.get(name, unknown)}) print(f 输入: {str(inputs)[:200]}) def on_chain_end(self, outputs, **kwargs): print(f 输出: {str(outputs)[:200]})然后在组合链路时给某个子链单独贴标签、单独传配置retrieve_chain ( RunnableParallel(...) | retrieve_prompt | llm | StrOutputParser() ).with_config({run_name: retrieve_answer, callbacks: [MyLogCallback()]})这种做法最大的价值在于生产环境出了问题你可以按run_name快速定位到具体是哪段 Runnable 出了状况而不是对着整条链路瞎猜。4.3 工程落地时的 RunnableConfig 参数传递提到配置就绕不开RunnableConfig。它是 LangChain 里传递运行时参数的标准容器常见用途包括传递 API Key、控制模型参数、指定回调处理器、设置标签等。怎么理解它的作用你可以把 Runnable 想象成一个函数config是它的隐藏参数。不需要每个 Runnable 都关心 config 里有什么只有用到的那一层才会读取。这样设计保证了管道组合的简洁性——前面提到的那个prompt | llm | parser如果你想让后面的模型用不同的 temperature不需要重写链条直接chain.invoke( {question: 讲个笑话}, config{temperature: 0.8} )但这个写法对有些组件是无效的因为temperature并不是标准 RunnableConfig 字段。正确的做法是提前用bind绑定模型参数llm ChatOpenAI(modelgpt-3.5-turbo, temperature0.7) chain prompt | llm | parser所以这里有个经验运行时动态变更的参数走config如 callbacks、metadata、run_name模型本身的关键参数走bind或初始化。混着用会踩不少隐形的坑。5. 排查 LCEL 链路的三种高效姿势5.1 姿势一拆链法——把管道拆开逐步执行管道写成长链后第一次报错往往让人头皮发麻——错误信息显示的是整个RunnableSequence报错却不知道具体是哪个环节。我自己的排查习惯是不要一上来盯着错误信息猜先把链子拆开手动一步一步执行每一环。比如prompt_val prompt.invoke({keyword: 智能手表}) print(prompt_val) # 看看提示词模板拼出来到底对不对 llm_output llm.invoke(prompt_val) print(llm_output) # 看看模型返回了什么类型、什么内容 final StrOutputParser().invoke(llm_output) print(final)这样做的好处是你能立刻确认问题出在提示词本身写得烂、模型没返回预期格式还是输出解析器解析失败。等定位到具体环节后再修复比对着整条链子瞎改要快得多。5.2 姿势二中间输出注入——用 print 或者回调拦截拆链法适合本地调试但有时候问题只有在整条链跑起来才复现比如某个中间变量只有在特定输入下才会异常。这时候我会临时在管道中间插一个打印节点from langchain_core.runnables import RunnableLambda debug_node RunnableLambda(lambda x: print(x) or x) chain prompt | llm | debug_node | parser这个调试节点的作用就是原样转发数据同时把数据打印到控制台。打印完之后再接回原来的链路。优势是极简缺点是不适合生产环境但本地调试足够好用。5.3 姿势三LangSmith / 本地 trace 兜底真正进了生产或者比较复杂的开发环境我还是建议接上可观测性平台。我记得 LangChain 生态里最常用的方案是 LangSmith官方出的。接口对接很简单设置环境变量后所有 Runnable 自动被追踪。你可以在平台上看到每一步的延迟、token 消耗、完整的输入输出链路图排查问题效率会高一个档次。如果不想把数据发到云端也可以找一些自托管的替代方案或者自己实现一个简单的存储型回调处理器写到本地日志或数据库表里。我不太建议所有项目一上来就接平台但一旦链条超过 5 个节点、开始有用户反馈答案不对却无法复现时trace 就是你唯一能抓着不放的救命稻草。6. 项目实战中反复踩到的几个坑6.1 类型注解错误导致 RunnableLambda 静默失败RunnableLambda包装函数时如果函数的类型注解不规范LangChain 有时候会做奇怪的类型转换。我遇到过一次函数接收的是一个dict我在类型注解里写成了str结果函数内部跑起来直接抛异常——但异常发生在链路的深处错误信息又很隐晦我足足花了半小时才定位到是这个注解问题。后面就长记性了被 RunnableLambda 包装的函数类型注解必须和实际输入输出一致。特别是使用config参数时第二个参数要显式标注为RunnableConfig否则某些版本下会被误判。6.2 链内可并行化判断的误解LCEL 的并行和异步并不是自动的。RunnableParallel只是把多个子 Runnable 组合成一个并行 runner但如果内部用的是同步调用它也不会真正并发执行多个模型请求。我在一个项目里天真地以为把三个链子包在RunnableParallel里三个模型调用就会自动并发。实际测试下来三个请求还是串行跑的因为默认用的是同步执行器。想真正并发需要from langchain_core.runnables import RunnableParallel import asyncio parallel_chain RunnableParallel( sentimentsentiment_chain, keywordskeyword_chain, should_replyreply_chain ) result asyncio.run(parallel_chain.ainvoke({comment: ...}))用异步接口ainvoke才会真正并发。所以做性能优化时一定要搞清楚能并行和真的并发执行的区别。6.3 不同版本的 API 兼容性问题LangChain 生态的版本迭代速度很快我踩过最典型的一个坑某个项目用的 langchain 版本里RunnableParallel的字段传参方式变了导致原来的链子直接报 TypeError。后面我把关键依赖的版本锁死升级前先在小环境里做回归测试才没有在线上翻车。我的建议学习 LCEL 时优先看当前稳定版本对应的官方文档如果搜索中文资料一定要留意文章发布对应的大版本。很多博客上讲的 API 在最新版本里已经改名或者删除了。6.4 生命周期钩子与流式输出的关系最后一个要提的坑和stream有关。我一直以为只要链条里的模型支持流式返回整条 LCEL 链就会自动变成流式输出。实际上中间的每个 Runnable 都得愿意配合。比如某个RunnableLambda包装的函数如果一次性输出了全部内容就会打断上游的流式传递。这也是为什么 LangChain 在设计上强调**流式传播需要整条链路的所有组件都支持**否则就会退化成等全部生成完再一次性吐出来。如果你的链子里夹带了不可流式的组件又希望整体保持流式可以考虑把流式部分拆成独立的 Runnable再在业务侧拼装。7. 一次 LangChain 版本大升级后的踩坑整理这里单独补一段我真实的迁移经历对很多已经从旧版走过来的开发者应该有共鸣。我有个线上问答服务早期用的是基于LLMChain的写法。去年年底某个版本的升级官方直接把旧版链子的一堆接口标记为 deprecated我对接的好几个代码片段开始刷警告。趁那次机会我把整个服务迁移到了标准的 LCEL 写法。迁移过程中最有趣的一个问题是旧的LLMChain里自带memory参数可以直接记住历史对话。换成 LCEL 以后这个能力没了得自己去外面包一层消息历史的 Runnable——或者改用RunnableWithMessageHistory。我第一次迁移时没意识到这个变化测试对话时发现模型完全失忆排查了很久才找到原因。如果你也在做类似的迁移给你列一个我当时的检查清单应该能少走点弯路迁移点旧写法LCEL 推荐写法提示词与模型组合LLMChain(llm..., prompt...)prompt | llm输出解析在 LLMChain 里传output_parser链尾部接StrOutputParser()多步骤流程SimpleSequentialChainRunnableLambda或RunnableBranch并行调用自己写并发逻辑RunnableParallel多轮对话记忆LLMChain 内置 memoryRunnableWithMessageHistory或外层消息管理部分变量填充chain.run(partial_variables...)prompt.partial(...)回调与追踪传 callbacks 到 chain通过 config 传 callbacks迁移完成后代码行数大概降了 30%可读性明显提升而且因为每个节点都是独立 Runnable加日志和追踪变得更顺手。8. 给正在学习 LCEL 的人几个真诚建议最后这部分算是我个人的体会不涉及太多技术细节但确实是在大量项目里砸出来的经验。第一不要试图一次读懂所有 Runnable 变体。先盯住invoke、batch、stream、bind这几个最核心的行为其他变体遇到了再学。Runnable 体系的设计初衷是少而精的组合原语越往后用你会发现大部分需求都只是这几个原语的组合很少需要真的自己写一个全新的 Runnable 类。第二先在 Jupyter Notebook 里做实验再上工程框架。LCEL 的声明式风格非常适合交互式开发。先确认某个链子在你想要的数据上能跑通再把它封装到 FastAPI 接口或者 Agent 工具里。直接上工程代码的话调试成本会高很多。第三时刻留意数据的类型和形状。Runnable 之间靠隐式格式转换衔接但你写代码时心里必须清楚这一步输入是什么str、dict、PromptValue 还是 Document输出是什么。很多 LCEL 链子跑不通不是逻辑错而是数据形状和预期不一致。第四多看看官方仓库里最新的 runnable 示例。LangChain 的文档和 GitHub 示例更新非常快社区里很多最佳实践其实已经过时了。以官方文档当前版本为准遇到矛盾时调试代码比找旧博客靠谱。第五组合超过 5 个节点时尽早接 trace。人脑能同时追踪的隐式状态很有限超过这个数量排查效率会断崖式下降。哪怕只是本地往日志里打一份也比全靠肉眼推理强。LCEL 这套体系初看确实有点陡尤其是习惯了命令式编程的人一开始会觉得不过就是个语法糖。但当你真正用它搭出一个带检索、带路由、带并行分支、带重试回退的生产级链路时就会明白——它是 LangChain 对语言模型应用开发这件事的一次重新抽象把原先散落在各个代码块里的流程控制收敛成了几个简单有力的组合原语。把 Runnable 弄明白之后你再看 LangChain 的其他文档会有一种横着看任何链路都是几个 Runnable 嵌套的通透感。
