做LangChain项目做到一定阶段你一定会遇到一个绕不开的问题单靠一次模型调用根本办不成事。拿最简单的自动翻译来说你直接问模型“把这句话翻成英文”它确实能回你一句英文但你要的是稳定的输出格式、要剥离它夹带的冗余解释、要能编程处理后续流程——这时候“链条Chains”和“LangChain Expression LanguageLCEL”就来了。前面几篇分别聊了提示词模板、模型接入和输出解析这些组件但光有它们就像手里攥着一堆零件得有一条流水线把它们装成能干活的东西。Chains做的就是这件事把若干步骤串联成一个处理链路前一步的输出自动变成后一步的输入。而LCEL作为LangChain主推的链条声明式写法用最直观的管道符语法把组件连起来。这篇作为进阶系列的第六篇我会从原理到实操把链条彻底讲透。没有基础也不用担心我尽量用直线逻辑推进先说明为什么需要Chain再拆解LCEL的语法和运行机制然后带你把三类常见链条从头搭一遍最后分享几个我实测中经常踩的坑。这篇文章之后你再回看LangChain的文档和社区代码会发现绝大多数示例都是LCEL在串场。1. 为什么需要Chain一次模型调用解决不了完整任务1.1 单次调用的真实局限很多人刚接触LangChain时会觉得“直接调模型不就行了吗要Chain干什么”。这种想法在Demo阶段完全没问题但稍微沾点实际业务就会露馅。我举个例子你想做一个“给代码自动写注释”的小工具。最朴素的实现是写一个提示词把用户代码塞进去问模型“请给这段代码补上注释”。这个方案至少有四个问题。第一提示词里必须写一大段固定指令这个指令在每个请求里都要重复传输浪费token也拖慢响应。第二模型回你的可能不止注释它会顺手解释代码逻辑、提醒注意事项这些附加值看着贴心但你的程序根本没法解析。第三你如果想输出成JSON或者带标记的结构化文本纯靠提示词约束是靠不住的模型一高兴就自由发挥。第四你如果想支持Java、Python、Shell多语言每个语言写一套逻辑整个调用代码会膨胀成一堆互相穿插的if-else。传统写法下这些“胶水逻辑”分散在业务代码里每次加需求就要跟着改一大片。链条和LCEL解决的就是这类结构性问题把“固定指令”“模型调用”“输出解析”“后续处理”都抽象成一个个独立的组件组件之间通过统一接口连接业务代码只需要关心数据流怎么走不用关心每个步骤内部怎么实现。1.2 链条在LangChain里的定位我们可以把链条想成一条工厂装配线。原始输入是工件提示词模板是第一个工位它负责把固定格式的指令和用户变量组装成一份完整的“订单”模型是第二个工位它根据订单完成推理输出解析器是第三个工位它把模型产出的粗糙回答打磨成规格统一的成品RunnableLambda之类的组件则是后面追加的质检和改模工位。每个工位做完自己那件事就把半成品交给下一个工位。这个类比里最关键的一点是方向性。链条之所以叫“链”是因为它严格按照顺序执行上一个组件的输出类型必须能被下一个组件的输入类型接住。LangChain在设计上把几乎所有东西都实现成了Runnable接口这就是LCEL能用管道符直接连接组件的前提。下面这张表列了链条里的几个常见“工位”你后面搭链时会反复遇到。组件类型工位角色常见类提示词模板把变量组装成模型真正读到的提示词PromptTemplate / ChatPromptTemplate模型封装完成推理请求ChatOpenAI / ChatOllama / ChatAnthropic 等输出解析器把模型回复转成字符串或结构体StrOutputParser / JsonOutputParser工具函数清洗、转换、路由RunnableLambda透传变量原样传递输入供分支或多个环节引用RunnablePassthrough并行组合同时执行多条子链后合并结果RunnableParallel早期LangChain里组装链条靠的是LLMChain、SimpleSequentialChain、SequentialChain这类显式类写起来绕调试也费劲。官方后来把声明式的LCEL定为主推方案原因就是它把“串联”这个动作简化成了视觉上一个管道符的事代码即流程图链条长了也不至于看到头晕。2. LCEL语法拆解管道符背后的Runnable协议2.1 管道符不是语法糖而是协议绑定第一次看到prompt | model | parser这种写法时多数人会以为管道符是某种重载运算符做了点语法甜头的封装。这种理解不全面。LCEL中的管道符之所以能把两个不同组件连起来是因为它们都实现或者兼容了LangChain的Runnable协议。这个协议规定了一套统一的方法核心有四个invoke处理单个输入、batch批量处理、stream流式输出、ainvoke异步处理单个输入。对应的异步版本就是ainvoke、abatch、astream。也就是说你写的prompt | model实际执行时相当于“先用prompt的invoke处理输入再把输出交给model的invoke”。因为两边都遵循同一个协议LangChain只需要按顺序调每个组件的invoke就能完成整条链的处理。这是它最大的价值组合后的对象本身依然是一个Runnable可以继续和其他Runnable相连接也能继续调用协议里的所有方法。所以你拼出来的链本身又能被别的链嵌套相当于组装好的模块还能继续当零件用。写一个简单例子感受一下。同一个链你可以同步调用、批量调用、流式调用from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser prompt ChatPromptTemplate.from_template(请用一句话解释{concept}) model None # 换成你实际可用的模型 parser StrOutputParser() chain prompt | model | parser # 单个调用 result chain.invoke({concept: 依赖注入}) # 批量调用返回多个结果 results chain.batch([{concept: 递归}, {concept: 闭包}]) # 流式调用逐段输出 for chunk in chain.stream({concept: 编译器}): print(chunk, end)不需要额外写任何循环或者分发逻辑协议把同步、异步、批量、流式这些能力都统一开放了。这就是LCEL最核心的设计意图——声明式组合不是省了代码量而是让组合后的东西成为一个能力完整的整体。2.2 RunnableLambda把普通Python函数包装成流水线零件实际链条里不可能只有LangChain自带的组件总有几段逻辑是需要你自己写的比如把用户输入里的空白字符清掉、把模型输出的文本切成列表、根据关键字判断下一步走哪条分支。这时候就需要RunnableLambda它可以把一个普通Python函数包装成Runnable组件让它能被管道符连接。from langchain_core.runnables import RunnableLambda def normalize(text: str) - str: text text.strip() return text.replace(\n, ) chain RunnableLambda(normalize) | prompt | model | parser这里有个实际使用中值得注意的点RunnableLambda包装的函数如果只有一个参数它会自动接收上一环节的输出如果你想传多个字段通常要把函数定义成接收一个dict参数名对应dict的keyLangChain才会自动从中取值。但这种约定有时候不够直观我更推荐显式点透传字段def extract_question(payload: dict) - str: raw payload.get(raw_text, ) return clean(raw) # 显式从dict里取字段避免依赖隐式绑定 chain RunnableLambda(lambda x: extract_question(x)) | prompt | model | parser2.3 RunnablePassthrough不做事也是重要的事链条里经常出现“这个输入要原样传给后面一个环节但同时也给旁边环节用一份”的情况。比如检索增强问答里用户的问题既要拿去查向量库又要在拼提示词时原样放进“问题”位置。前者有retriever处理后者你不想做任何加工。RunnablePassthrough干的就是这个活输入是什么输出就是什么原封不动地透传。这个名字容易让人觉得它“什么都没做”但在并行分支里它是让变量按需求分配到不同路径的关键。后面搭检索问答链时你会看到它的实际用途。它也是LCEL里最容易被忽略但几乎每个真实链都会碰到的组件之一。3. 用LCEL搭建三条典型流水线3.1 最简LLM链路Prompt Model OutputParser先从最小可运行的链条开始。假设我们要做一个“用一句话解释某个概念”的功能直接把用户给的词填进模板交给模型最后把模型的文本回复收成纯字符串。代码长这样from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser from langchain_openai import ChatOpenAI prompt ChatPromptTemplate.from_template( 你是一名资深工程师。请用一句话解释「{concept}」只说结论不要加任何前缀。 ) model ChatOpenAI(modelgpt-4o-mini, temperature0) parser StrOutputParser() chain prompt | model | parser result chain.invoke({concept: 依赖注入}) print(result)这条链能跑起来的底层逻辑是三个组件之间输入输出类型正好匹配prompt.invoke返回一个PromptValue模型能接收PromptValue并返回AIMessageStrOutputParser专门负责从AIMessage里把文本内容提取成str。类型匹配是LCEL能否串联的硬规则这点请先记住后面章节的排错内容会反复提及。顺便说一句模型接入方式。上面用了ChatOpenAI实操中完全可以根据环境换成别的提供商只要它遵循LangChain的模型封装接口即可。这里不建议在初始化时写死太多参数先让链路跑通再去调temperature和max_tokens这些细节。3.2 检索问答链让模型带着资料回答问题前面这位是最简链路但它有个天然缺陷模型的知识止于训练截止时间对于私有资料、实时变化的内容完全没法答。典型场景是给公司内部文档做问答机器人模型不该凭空编造答案而应该先检索相关文档片段把片段塞进提示词参考资料里再让它基于这些资料回答。用LCEL写这个链的核心是输入重构。链的输入是用户的一个问题但提示词模板需要两个变量一个是检索出的上下文片段一个是用户问题本身。所以我们先用一个dict做并行映射让retriever去查资料同时让RunnablePassthrough把问题原样透传from langchain_core.runnables import RunnablePassthrough retriever vectorstore.as_retriever(search_kwargs{k: 4}) prompt ChatPromptTemplate.from_template( 以下是参考资料\n{context}\n\n请根据资料回答问题{question} ) rag_chain ( {context: retriever, question: RunnablePassthrough()} | prompt | model | parser ) answer rag_chain.invoke(我们的服务出现502应该怎么办)这里{context: retriever, question: RunnablePassthrough()}就是一个RunnableParallel的简写形式。LCEL发现外层是dict会自动把每个value当成一个子链去执行执行完了把结果按key合并成一个新的dict作为下一步prompt的输入。retriever接收到的输入是这条链传进来的原始参数也就是用户问题。这属于LCEL里非常经典的“输入重构”套路你在社区看到的检索增强问答几乎都是这个结构。为什么非要这么绕一圈直接先把问题存成一个变量不行吗关键在于LCEL的链是声明式的中间没有显式的变量赋值语句你想让同一个原始输入同时供给两个环节就得借助RunnableParallel和RunnablePassthrough组成一个“输入分发器”。这是理解LCEL写法的分水岭越过了这一关后面几乎所有复杂链你都能看懂。3.3 多步处理链先拆解、再查询、后汇总检索问答解决的是“资料引入”还有一种常见需求是“多个模型调用顺序接力”。比如做一个多语言摘要功能用户输入中文链先把它翻译成英文再用英文做一句话摘要最后把摘要翻回中文。每一步都是一个完整的“提示词模板模型输出解析器”子链子链之间通过字符串传递。translate_en ChatPromptTemplate.from_template( Translate the following Chinese text into English:\n{text} ) | model | parser summarize_en ChatPromptTemplate.from_template( Summarize the following text in one sentence:\n{text} ) | model | parser translate_zh ChatPromptTemplate.from_template( Translate the following English summary into Chinese:\n{text} ) | model | parser chain translate_en | summarize_en | translate_zh result chain.invoke({text: 机器学习模型在训练过程中需要大量标注数据这往往成为项目落地的主要成本瓶颈。})这个例子最能说明类型匹配的重要性。每一步如果不加输出解析器传入下一环的就是AIMessage对象而ChatPromptTemplate期待的是文本变量组装提示词时就会出问题。加上StrOutputParser之后链里流动的始终是普通字符串各环节才能顺利衔接。我见过不少新手在这里卡住报错信息里写着“无法把AIMessage格式化进模板”其实就是缺了一个解析器或转换环节。4. 并行、分支与合并从“流水线”到“车间网络”4.1 RunnableParallel一次请求同时跑多条链实际业务里经常出现“一条链路里需要从多个角度同时处理同一份输入”的情况。比如写文案工具用户给一段产品介绍你希望模型同时输出“营销版本”和“用户群定位分析”两个结果这两个任务相互独立完全没有先后依赖。串行执行不仅慢还会在代码里额外引入中间变量。RunnableParallel就是为此准备的。它的用法很直观把多个子链以关键字参数形式传进去执行时并行跑最后把每个结果按key收进一个dict里返回。from langchain_core.runnables import RunnableParallel marketing_chain ( ChatPromptTemplate.from_template(根据这段产品描述写一条朋友圈文案{text}) | model | parser ) position_chain ( ChatPromptTemplate.from_template(分析这段产品描述的目标用户画像{text}) | model | parser ) parallel_chain RunnableParallel( marketingmarketing_chain, positionposition_chain, ) result parallel_chain.invoke({text: 一款支持一键导入备份的网盘工具}) print(result[marketing]) print(result[position])这里要提醒一句LCEL的“并行”更适合模型调用这类网络IO密集型的任务它会把子链们调度成可等待的异步任务同时发起请求总耗时会比串行明显缩短。但如果你的子链里有大量本地CPU计算依赖这种方式并不能获得真正的多核并行加速那是另一个范畴的问题了。4.2 RunnableBranch让链自己决定走哪条路链总是按顺序执行还不够有些场景要求在链条开始前根据输入内容做一次路由。最典型的是用户输入如果是技术问题就走检索问答链让模型尽量引用资料如果是闲聊就走普通对话链回复轻松一点如果命中敏感词直接返回固定话术连模型都不用调。RunnableBranch就是为了这个场景设计的。它的写法是一组“条件函数分支链”的配对最后加一个默认链跟编程里的if-elif-else结构几乎一一对应from langchain_core.runnables import RunnableBranch def is_coding_question(text: str) - bool: return 报错 in text or 代码 in text or 怎么实现 in text tech_chain coding_prompt | model | parser chat_chain normal_prompt | model | parser fallback_chain RunnableLambda(lambda x: 这个问题我先记录下来稍后人工确认。) chain RunnableBranch( (is_coding_question, tech_chain), chat_chain, # 默认分支 ) result chain.invoke(我的Python代码报错KeyError怎么办)执行时LCEL会先调用条件函数拿到布尔值。为True就走对应分支链全部不满足就走最后的默认分支。默认分支是必填的否则会报缺少路由终点。这个结构在纯Python里写其实也就是几行if-else的事但放进了LCEL之后你就可以把整个分支选择逻辑挂在链条内部外部调用方永远只需要chain.invoke(...)不需要关心路由细节整体封装性好了不止一个级别。4.3 实用组合并行加分支的完整链路单用RunnableParallel或者单用RunnableBranch都只能算入门把它们一把梭才是真实项目的常态。举个例子一条“智能客服”链用户问题进来先路由判断是不是技术问题走技术分支时并行做两件事——查知识库拿上下文、让模型判断问题是否紧急然后合并结果再统一拼接成最终回复。链路画出来大概是这样的逻辑输入问题 - 路由 - 技术分支检索上下文 紧急判定并行执行 - 汇总提示词模板 - 模型 - 解析输出。retriever vectorstore.as_retriever(search_kwargs{k: 3}) qa_prompt ChatPromptTemplate.from_template( 资料{context}\n用户问题{question}\n紧急程度判断{urgency}\n请据此给出解决方案。 ) def check_urgency(text: str) - str: if 紧急 in text or 宕机 in text or 无法访问 in text: return 紧急 return 普通 tech_entry RunnableParallel( contextretriever, questionRunnablePassthrough(), urgencyRunnableLambda(check_urgency) | (lambda x: x), # 这里简化处理实际可以单独让模型判断 ) tech_chain tech_entry | qa_prompt | model | parser这个例子的重点不是代码本身而是你会发现LCEL组合出来的链条依然可以被继续拼接。可以是分支的子链、可以是并行的某个分支、也可以是整条链再作为某个更大链条的一个环节。这种“万物皆Runnable、拼接再拼接”的能力才是它作为LangChain底层表达语言的底气。5. LCEL实测中容易踩的坑与调试技巧5.1 流式输出为什么突然失灵LCEL支持流式输出这是很多人选它的重要原因。但流式输出有一个隐藏前提链条中所有环节都得支持流式或者是纯透传的。一旦你在模型输出之后插了一个RunnableLambda做文本处理比如清洗、截断、二次调用别的接口那么整个链的流式就会“退化”——不再逐token输出而是等这个非流式步骤处理完一次性吐出一个完整片段。我实测过的典型场景是一个问答链模型输出后想用RunnableLambda过滤掉敏感词原本好好的逐字输出变成了卡顿一下然后整段出现。解决思路有两个如果过滤逻辑可以不参与流式就把它放到链末端之后用普通的result chain.invoke(...)再处理如果确实要边生成边过滤那得在stream循环里逐chunk处理而不是把它塞进链条中流。# 正确的流式处理方式 for chunk in rag_chain.stream({question: 你们的服务SLA是多少}): print(chunk, end, flushTrue) # 错误的做法在chain里插一个过滤节点后直接stream # chain model | RunnableLambda(filter_sensitive) | parser5.2 invoke参数类型与变量数量不匹配这条坑几乎人人踩过。LCEL里的提示词模板定义了几个变量调用链条时就得传够几个值。模板里写了{context}和{question}你只传一个字符串进去系统不会自动帮你拆而是直接报“缺少key”之类的错误。这不是LangChain设计笨而是它不想替你偷偷猜测输入应该怎么映射到模板变量上。我的建议是给链条加一个薄薄的包装函数把调用入口收窄。业务代码那边永远只传一个question字符串进来函数内部去构造完整的参数字典。def run_rag(question: str) - str: return rag_chain.invoke({question: question})这样链条需要的变量怎么变化受影响的地方只在这个函数内部调用方完全无感。我实践中发现这一步对团队协作尤其重要不同人写出来的调用风格五花八门统一入口能省掉大量沟通成本。5.3 中间结果看不见给流水线加“观察窗”LCEL管道写法的一个副作用是链条一旦执行中间环节的输入输出就被吞掉了你只能看到最终结果。早期排查问题如果最终答案不对很难判断是检索拉错了资料、模型理解偏了还是解析器格式不对。我自己调试时习惯临时在关键节点插一个观察用的RunnableLambda把经过的数据打个标记后再原样传出。这种“观察窗”不会改动数据但能让你清晰看到每个环节的输入输出形状。def inspect(step_name: str): def _inner(x): print(f[{step_name}] 收到数据:, x) return x return RunnableLambda(lambda x: _inner(x)) chain ( {context: retriever, question: RunnablePassthrough()} | inspect(检索后) | prompt | inspect(组装提示词后) | model | inspect(模型输出后) | parser )生产环境记得把这些观察窗全部删掉否则每次请求都会多打一堆日志数据量大时还挺影响性能。另外这类观察窗还能用来验证类型匹配问题比如看看进入下一个提示词模板的到底是字符串还是AIMessage这比对着报错猜测要高效得多。5.4 版本迁移从旧式Chain类到LCEL需要注意什么如果你翻的是比较早的LangChain教程会看到大量from langchain.chains import LLMChain、SimpleSequentialChain、SequentialChain的写法。这类类式链条在早期确实是主力但如今官方已经把LCEL作为主推方向新项目再按老写法搭链有几个实操上的不便利一是老式类链的参数传递方式更隐晦跨步骤传值经常要靠input_key和output_key层层声明链条一长就很难理清二是代码仓库迭代过程中部分老链类所在模块的位置会变动升级依赖版本时容易碰到导入被移除的问题三是老链条对流式、批量、异步的支持明显不如Runnable协议下的LCEL来得统一。我的建议是读老教程时把重点放在理解它的设计思想——哪一步该做什么、变量怎么流转——然后尽量用LCEL的管道符重写它的示例。现在LangChain生态里LCEL已经成了跨版本最稳定的表达方式你花半小时熟悉管道符语法换来的是后面所有项目的通用能力。如果你在维护历史代码短期内可以继续跑老链但新功能一律用LCEL写慢慢把老链路替换掉这样过渡成本最低。版本迭代还没停但LCEL的思路值得先吃透我自己把LCEL用得顺手之后最大的体会是写链之前先别急着敲代码拿一张纸把“数据流图”画出来。每次写入方向从这里走、哪个环节产生什么类型的输出、下一步组件期望什么类型的输入把这张图列清楚了LCEL代码几乎是一气呵成的。它容易让人忽略的反而是数据类型的流转而这点正是调试时最花时间的地方。如果你刚开始接触LangChain的链条可以拿3.2节的检索问答链练手把RunnablePassthrough、RunnableParallel、RunnableLambda都拆开试一遍。这个链路是LangChain黄金组合覆盖了LCEL的绝大多数核心用法。等你把这条链吃透再去看社区里各种花哨的实现会发现内核大同小异——不过都是把流水线拆得更细、并得更巧罢了。