LangChain Chains核心指南：用LCEL构建可复用的AI工作流编排

玩LangChain到一定阶段很多人都会卡在同一个地方单次调用大模型已经没问题了Prompt模板也会写了但一碰到稍微复杂的任务——比如先提取关键词再生成大纲最后补全文代码就开始失去控制要么到处塞中间变量要么整个流程拆成好几个函数硬拼。这时候就该正式认识一下Chains模块了。简单说Chains解决的是这样一件事你怎么把多个步骤组织成一条可复用、可观察、可替换的工作流水线。在LangChain当前的版本语境里这个组织的核心已经不完全是早期的Chain类而是LCEL这套表达式语法加上一批编排组件。这篇模块二就把Chains的工作流编排原理、常见模式和我实际跑过的代码一次讲透。适合谁看已经写过几段LangChain基础代码、但对任务编排还停留在if-else阶段的开发者。我把新老两套体系的关系、五种编排模式、完整示例和生产化检查点都串起来讲争取你看完能直接把项目里的大模型调用堆改造成正经的工作流。1. 理解Chains以前先搞清楚它在LangChain里管的到底是哪摊事1.1 四个模块的定位模型解决会说话Chains解决怎么组织LangChain的命名经常让人误会以为它只是个大模型封装库。实际上它把能力拆成了四个互相咬合的模块Models负责接入和调用大模型统一了对话、文本补全、嵌入等接口。Prompts负责管理Prompt模板解决怎么问的问题。Chains负责把模型调用、函数调用、检索步骤编排成一条有序的流水线解决怎么组织的问题。Agents负责让模型自己决策下一步该调哪个工具解决怎么做选择的问题。我的理解是Models和Prompts是零件Chains是把零件装配成产线的机械臂Agents则是产线上那个会看情况改工艺的老师傅。没有Chains你只能一次一次手动拼装零件有了Chains你才能定义一条可重复执行的生产线。1.2 用工厂流水线来理解Chain到底是什么想象一家组装工厂零件进来A工位加工B工位检测C工位包装最后出货。每个工位只负责一件事但整条线是一个整体——你给它一个原材料它给你一个成品。Chains就是这个抽象的流水线不过每个工位是一个可执行的节点它可能是一次模型调用LLM/ChatModel一个普通Python函数通过RunnableLambda包装一个Prompt模板ChatPromptTemplate一个输出解析器StrOutputParser、JsonOutputParser等另一个子Chain一个节点接受某种输入产生某种输出后一个节点把前一个节点的输出继续加工。这种前一个输出喂给后一个输入的接力结构就是Chain最基本的模样。至于中间要不要保存历史、要不要分流、要不要根据条件走不同分支都是在这个基础上扩展出来的花样。1.3 版本演进之后Chain这个词的含义已经变了这是新手最容易懵的地方。在较早版本里LangChain定义了一批具体的Chain类比如LLMChain、SimpleSequentialChain、SequentialChain、RouterChain。代码风格是这样的先用一个类把Prompt和Model绑在一起再用另一个类把多个Chain串起来配置input_variables和output_variables靠字符串名对应。到了现在的主流版本官方主推的是LCELLangChain Expression Language。你不再需要继承Chain类而是直接用管道符把Runnable对象拼起来chain prompt | model | output_parser这个链的左侧是PromptTemplate中间是模型右侧是解析器。|运算符把前一个Runnable的输出作为后一个Runnable的输入最终得到一个全新的Runnable。所以现在聊Chains更准确的说法是一切遵循Runnable协议的组件通过组合形成的工作流。老的Chain类并没有完全消失Agent内部还有但新项目直接上手LCEL是更顺的选择。这个新旧取舍我在下一节细说。2. 新老Chain体系并存LCEL与传统Chain类到底该选谁2.1 传统Chain类解决过的问题以及它们现在的位置早期版本里几个经典Chain类各怀绝技LLMChain把PromptTemplate和Model绑定成一个可直接调用的单元。现在用prompt | model就能替代没必要再包一层。SimpleSequentialChain只适合前一个输出直接就是后一个输入的直线流程一旦中间要插别的变量就抓瞎。SequentialChain通过input_variables和output_variables手工声明哪些变量传给哪一步配置多、容易写错但能处理多输入多输出。RouterChain / MultiRouteChain根据一个路由规则把输入分发给多个子Chain这就是路由模式的雏形。这些类最大的问题不是功能不对而是可组合性太差。你想在中间插一步格式化结果、想在某个环节分支、想流式输出中间过程老的一套做起来很麻烦——要么继承重写要么硬塞进某个Chain类的参数里。代码一多链条变成一张靠字符串变量名维系的蜘蛛网改一个名字全盘崩。2.2 LCEL的Runnable协议为什么能统一一切LCEL做的第一件事是定义了一个底层协议Runnable。任何对象只要实现了invoke、batch、stream、ainvoke这些方法它就是一个Runnable。把多个Runnable用|连起来得到的还是Runnable。这个设计带来的几个直接好处接口统一模型调用、函数、解析器、检索器、整条链用法完全一样都是chain.invoke(...)。中间可观察链的任何位置都能挂回调你能看到每一步的输入输出。流式支持天然只要底层模型支持整条链可以做流式输出而老Chain类通常是全量等待结果。异步和平行.ainvoke()、.batch()、RunnableParallel都是协议自带的能力不用额外开发。还有一个很实在的特点LCEL的链是lazy的。你拼出来的只是描述真正执行发生在调用invoke的那一瞬间。这也让链本身可以像普通对象一样被传入、被返回、被嵌套。2.3 我的选型建议这几个月我在不同项目里两种都试过结论比较明确新代码默认LCEL老Chain类只在两种情况里保留——你要读别人写的旧代码或者你在Agent内部做定制因为AgentExecutor内部大量依赖旧的Chain/Action概念。维度传统Chain类LCEL组合式Chain学习曲线偏陡类多、配置项多平缓一条prompt | model | parser起步灵活度低靠字符串变量名接线高可随意插节点、分支、并行流式输出基本不支持天然支持模型配合即可调试友好度中间状态不好拿回调、逐层替换都很容易异步/并发支持有限.ainvoke()、.batch()内置适合场景维护旧代码、Agent内部一切新功能开发如果你是刚入门直接学LCEL完全不用回头啃老的SequentialChain那一套。只要把RunnablePassthrough、RunnableParallel、RunnableLambda、RunnableBranch这几个组件用熟就能覆盖绝大部分编排需求。3. 五种核心编排模式拆解顺序、并行、路由、条件、循环3.1 顺序模式最经典的管道直连顺序模式是最基础的编排适用于产出恰好是下一步的输入的流程。最典型的写法from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser summary_prompt ChatPromptTemplate.from_template( 用三句话概括以下材料的核心观点\n{content} ) writing_prompt ChatPromptTemplate.from_template( 根据下面的摘要展开一篇短文要求逻辑连贯、结构完整\n{summary} ) chain ( summary_prompt | get_chat_model() | StrOutputParser() | writing_prompt | get_chat_model() | StrOutputParser() ) result chain.invoke({content: 这里放原始材料})注意中间那个| StrOutputParser() | writing_prompt是怎么连接的模型的输出是字符串正好作为writing_prompt的输入变量的值。如果你需要把多个值传给下一步就不能这样直连要先动手组装dict。顺序模式最容易出错的地方也在变量对接。比如第二步的Prompt模板里需要的变量是summary它会在输入里找summary这个键——一旦上一步输出的不是字符串而是包含别的结构的对象这里就断掉了。3.2 并行模式一批独立任务同时跑再汇总结果任务之间互不依赖时可以并行执行而不是一个个排队省时间也符合直觉。LCEL里的实现方式是RunnableParallelfrom langchain_core.runnables import RunnableParallel first_step RunnableParallel( keywordkeyword_prompt | get_chat_model() | StrOutputParser(), outlineoutline_prompt | get_chat_model() | StrOutputParser(), ) compose_chain ( first_step | compose_prompt | get_chat_model() | StrOutputParser() )这里的first_step接收一个输入dict然后同时跑keyword子链和outline子链等两个都完成后把结果按key拼成一个新的dict{ keyword: 关键词结果, outline: 大纲结果 }这个合并后的dict再喂给compose_prompt里对应的变量。如果你想保留原始输入里的某个字段可以在RunnableParallel里加一项RunnablePassthrough()它会把上游输入原样拷贝一份放进结果里。这个技巧几乎是所有复杂链的必备很多变量莫名丢失的问题都是因为忘了这一行。3.3 路由模式按输入内容分发到不同子流程真实业务里单一流水线往往不够用。比如用户发来一段文本你先判断它是待翻译还是待总结再决定走哪条子链。这就是路由。传统路由要靠RouterChain那一套配置LCEL里用RunnableBranch就能实现类似逻辑而更灵活的做法是先用一次模型调用做意图分类再按返回结果决定走哪个分支。示例from langchain_core.runnables import RunnableBranch translate_chain translate_prompt | get_chat_model() | StrOutputParser() summary_chain summary_prompt | get_chat_model() | StrOutputParser() # 先用分类Prompt让模型输出 translate 或 summary route_chain ( RunnablePassthrough.assign(routeclassify_prompt | get_chat_model() | StrOutputParser()) | RunnableBranch( (lambda info: translate in info[route].lower(), translate_chain), (lambda info: summary in info[route].lower(), summary_chain), lambda info: 无法识别请求类型请重新输入。 ) )RunnablePassthrough.assign做了件很关键的事它把上游输入dict复制一份额外加一个route字段。这样后面的RunnableBranch既能拿到原始内容又能拿到分类结果不会因为中间步骤改了数据结构而丢数据。RunnableBranch的每个分支是一个元组第一个元素是判断条件的函数返回True就走对应子链否则继续往后匹配。最后一个参数可以是兜底逻辑——既可以是链也可以是普通函数。我在实际项目里发现路由结果常常不干净。模型返回的可能是translate!或者Translate所以判断时建议用in加lower()而不是做完全等于比较。3.4 条件分支不靠模型靠规则决定走向不是所有的分支都需要模型来决策。有些场景规则非常明确比如输入为空直接报错长度超过500走长文本处理链这时候让模型跑一趟纯属浪费token。条件分支的惯用思路是先用RunnableLambda写一个Python函数在里面做判断动态返回对应的Runnablefrom langchain_core.runnables import RunnableLambda def route_by_len(info): text info[text] if len(text) 500: return long_text_chain return short_text_chain branch_chain RunnableLambda(route_by_len) | RunnableLambda( lambda runnable: runnable # 只是为了演示RunnableLambda可以返回一个Runnable )这里有个容易搞混的点RunnableLambda的函数里如果返回的是一个Runnable对象LangChain会自动去调用它并把结果作为下一段的输出。所以你可以把规则判断和子链选择放在同一个函数里返回选中的chain对象就行。不过这块写多了你会发现规则分支和路由模式界限并不绝对。我的建议是只要判断规则能用Python写清楚就不要调用模型模型只在判断没有明确规则、必须靠语义理解时才上场。省token是一方面更重要的是确定性——规则的稳定性远高于模型输出。3.5 循环模式让流程自己决定要不要再来一轮在Agent场景里链会循环执行多次模型决定调用工具工具返回结果模型再判断是否完成任务。这就是ReAct里思考-行动-观察的迭代循环。严格说循环逻辑通常写在AgentExecutor这类执行器里而不是靠LCEL的运算符。但你仍然可以在自建工作流中模拟循环from langchain_core.runnables import RunnableLambda def loop_until_done(info, max_rounds3): current info for i in range(max_rounds): result step_chain.invoke(current) if result.get(done): return result current result return {error: f未在{max_rounds}轮内完成} loop_chain RunnableLambda(loop_until_done)这种模式适合那些模型可能需要多轮修正才能得到合理答案的任务比如代码调试、格式修正。但有几个点必须注意一定要设最大轮数防止死循环烧光token。每轮之间要传递状态别让上一轮的结果丢失。循环内的每一步尽量有独立的超时控制。4. 完整示例把一份多步骤写作任务做成一站式工作流4.1 需求拆解从一个粗糙想法到四个明确节点为了让你直观看到上面的模式怎么组合我搭一个完整的应用输入一个主题词自动产出一篇结构完整的短文。流程拆成四步资料要点生成根据主题词生成主题下的3个核心要点。文章大纲生成基于要点生成带标题层级的大纲。分章节写作按大纲逐节写作这里简化成写一整篇。汇总润色对全文做最后的润色和去重。这四个节点里第1步和第2步是严格的顺序依赖没有要点就没有大纲。第3步依赖大纲第4步依赖第3步的产出。整体是一条直线但第1步我故意让它并行出三个方向的要点演示一下并行组件。4.2 代码实现LCEL把这些节点拼起来from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser from langchain_core.runnables import RunnableParallel, RunnablePassthrough point_prompt ChatPromptTemplate.from_template( 围绕主题{topic}从{angle}角度给出3条有信息量的要点各不超过30字。 ) outline_prompt ChatPromptTemplate.from_template( 根据以下要点生成一篇文章大纲要求层级分明\n{points} ) write_prompt ChatPromptTemplate.from_template( 根据以下大纲写一篇完整的短文正文不要出现大纲字样\n{outline} ) polish_prompt ChatPromptTemplate.from_template( 润色下面的文章使其表达更流畅、结构更清晰。只输出润色后的文章\n{article} ) point_parallel RunnableParallel( backgroundpoint_prompt | get_chat_model() | StrOutputParser(), marketpoint_prompt | get_chat_model() | StrOutputParser(), riskpoint_prompt | get_chat_model() | StrOutputParser(), ) chain ( {topic: RunnablePassthrough()} | RunnableParallel( topiclambda x: x[topic], pointspoint_parallel, ) | RunnablePassthrough.assign( pointslambda x: \n.join(x[points].values()) ) | RunnablePassthrough.assign(outlineoutline_prompt | get_chat_model() | StrOutputParser()) | RunnablePassthrough.assign(articlewrite_prompt | get_chat_model() | StrOutputParser()) | RunnablePassthrough.assign(articlepolish_prompt | get_chat_model() | StrOutputParser()) )第一行{topic: RunnablePassthrough()}是为了构造一个输入dict。后面每个RunnablePassthrough.assign都做一件事保留已有字段给dict加一个新字段。这种链式assign的方式非常建议新手复用——比你自己手工维护中间变量dict要稳得多。4.3 运行效果和几个值得留意的细节调用只需要一行result chain.invoke({topic: 智能家居}) print(result[article])执行后points字段里装的是三个角度各自的要点文本合并成多行字符串后传给outline_promptoutline生成大纲后再传给write_prompt最后polish_prompt润色。整个流程里topic这个原始字段通过第一层RunnablePassthrough一直保留到了最后——这就是为什么要保留原始输入的原因后面有任何节点想引用topic都能拿到。这里有个性能上的取舍三个方向的要点是并行跑的但如果你的模型API本身有并发限流并行反而会触发限流错误。生产环境里我一般会把并行的并发数调低或者干脆顺序执行。并行省时间是理论上的实际要配合限流策略看。另一个细节我在makechain里每一步都用assign去追加字段而不是重建dict。这样做的好处是每个中间步骤都仍然包含全部历史上下文方便调试时看某一环节的输入是什么。代价是多占一点token但对大多数应用来说可接受。4.4 扩展思路这个流程可以怎么升级这套骨架一旦建立扩展方向就很自由把write_prompt改成循环模式先写第一节、再写第二节最后拼接避免一次写太长。在polish之前加一个判断节点如果文章长度超过阈值才走润色否则跳过。接入记忆组件让链记住用户之前的修改意见。把chain封装成函数按主题批量生成内容配合batch()并行处理多个topic。5. 状态传递、记忆与中间结果检查调试Chain的三个基本功5.1 输入输出的传递逻辑你手写的dict就是链的全局变量LCEL的链里流转的核心是一个dict。每一步的Runnable接收这个dict处理后返回一个新dict或值。理解这条规则几乎所有变量传递问题都能迎刃而解。如果某一步返回的不是dict而是字符串比如StrOutputParser那下一步的Prompt模板会自动把它作为主变量填入。你可以认为LangChain做了隐式包装。如果某一步返回dict那所有字段都会成为下一步模板的候选变量。RunnablePassthrough通过assign或直接原样传递是最可靠的保留历史字段的方式。有个反直觉的点链里某个Prompt模板用不到某个字段并不代表这个字段会消失。除非某一步明确返回新dict覆盖了它否则字段会一直跟着走。所以调试时如果发现下游变量凭空多出或消失重点查中间哪一步重建了dict。5.2 记忆组件接入的常见位置长对话场景里Chain需要记住之前聊了什么。这一步通常不放在链中间而是在调用链之前维护一个外部消息列表每次invoke前把历史拼进Prompt模板的变量里。比如一个带历史摘要的工作流messages [] # 存历史消息 def chat_with_history(user_input): history_text format_history(messages) response chat_chain.invoke({ history: history_text, user_input: user_input, }) messages.append({user: user_input, assistant: response}) return responseLangChain本身也提供了memory类和专门的LCEL Runnable来管理历史比如RunnableWithMessageHistory。但说实话小项目里自己维护消息列表完全够用等真需要跨请求持久化时再去研究专门的记忆方案也不迟。5.3 中间结果观察挂回调还是逐段打印调试Chain最朴素也最有效的办法拆分执行逐段打印。把长链断成三四截每截的输入输出打印出来一眼就能定位是哪一步出了问题。想要不破坏链本身可以用回调。LangChain的BaseCallbackHandler里有on_chain_start、on_llm_end等钩子你可以把中间过程输出到日志里。更轻量的做法是在链里插一个RunnableLambda当探针from langchain_core.runnables import RunnableLambda def probe(label): def _probe(data): print(f[{label}] type{type(data)}, value{data}) return data return RunnableLambda(_probe) chain prompt | model | probe(after_model) | parser注意探针函数必须把输入原样返回否则等于往链里硬塞了一个改变数据的节点。如果项目里需要面向全链路的追踪LangChain的实验性trace工具也会收集步骤耗时、token用量和中间结果这种工具适合调优分析不适合日常极简调试。5.4 流式输出与中间状态的兼容问题LCEL支持链的.stream()但有个非常容易踩的坑只有所有节点都支持流式整个链才能流式输出。一旦链中间混入一个不支持流式的节点比如某些普通函数包装的RunnableLambda流式就会退化回全量等待。如果你做的是面向用户逐字显示的功能要对流式的退化有心理预期最好在链条设计阶段就让所有节点尽量保持可流式。6. 生产环境落地前的检查清单与高频踩坑实录6.1 错误处理与超时重试模型调用是最不稳定的环节链中任何一个节点抛异常整个invoke都会失败。最常见的是模型API的网络超时、限流、返回不符合解析格式。生产环境里我习惯做两件事第一给每个重要的LLM调用节点加超时和重试机制。最简单的办法是用Python的tenacity库包一层调用函数或者在模型客户端层面配置重试策略。核心原则是重试要有上限超过上限直接让整条链失败而不是无限等下去。第二为输出解析器做兜底。StrOutputParser几乎不会失败但JSON解析器一旦遇到模型返回多余的文字就会崩。兜底方案是加一个RunnableLambda做修复JSON把模型输出里的代码块标记剥掉再尝试解析def safe_json_parse(text): cleaned text.strip() if cleaned.startswith(): cleaned cleaned.strip() cleaned cleaned[cleaned.index({):] return json.loads(cleaned)这里值得多说一句模型输出不稳定是常态不是异常。设计链条时每个解析节点都要假定模型输出可能是脏的而不是假定它会乖乖输出规范JSON。6.2 并发与速率限制batch不是免费的午餐.batch()能同时处理多条输入看起来效率很高但你可能忽略了上游API的QPS限制。我曾经用一个8并发batch一次性跑几十条文本结果触发限流整批任务退化成一个个重试比顺序跑还慢。保险做法是给batch加并发上限或者在调用链的对象外面做信号量控制。更稳妥的方式是先用小批量测出API的容忍并发数再决定batch大小。另外链里有并行分支时分支数量也要算进并发别只看业务请求数。6.3 缓存与幂等同样的输入别让模型跑两遍很多链是同输入同输出的纯函数式流程。这种链非常适合加缓存直接用langchain_core.cache里的缓存组件或者更简单——自己在外部用字典/Redis以输入参数做key存最终结果。注意缓存key的设计如果链里包含随机性比如Prompt里写了用多样化的方式表达缓存就不该命中如果链中包含当前时间、用户ID这类动态字段要把它们排除在key之外。我踩过的一个坑就是缓存key包含了整个输入dict里面有时间戳字段结果缓存一次都没命中过。6.4 五个高频踩坑与排查思路坑现象根因排查方向变量名不匹配链跑起来报KeyError某步dict的字段名和模板变量不一致打印每步输出dict核对变量名JSON解析失败解析器抛异常模型在JSON外面加了说明文字加脏数据清洗层用容错解析上下文丢失后一步不知道前一步结果中间某步返回了新dict覆盖旧字段检查是否用了RunnablePassthrough保留历史流式失效用户端逐字效果没了链内混入不支持流式节点逐段测试哪些节点支持流式模型API限流报429或大量超时并行分支、batch并发过高降低并发确认API限额加重试策略这些坑有一个共同点它们几乎都不会在单段链的测试里暴露往往是在链变长、分支变多之后集中爆发。所以排查思路一般是这样构造最小复现输入把长链拆到只剩出错的那一段。在每段前后打印输入输出找到第一个不符合预期的节点。修复后逐段拼回去确认没有隐藏的变量覆盖。最后跑一遍完整链用真实输入的边界值测一遍。6.5 我建议的制作顺序新建一个工作流项目时别急着上大而全的编排。我的习惯是第一版只做一条最简直线输入 - Prompt - 模型 - 输出。跑通后再开始加节点加一个验证一个中间用探针打印观察。等节点数超过三个再引入并行、路由这些进阶组件。这样做的好处是每个环节的问题都在它刚出现时被解决而不是等到链条复杂到看不懂的时候才去排查。另外链的设计要尽量做成纯函数给定同样的输入产出可预期的结果。不要在链节点里偷偷改外部状态——虽然RunnableLambda里可以写任意代码但这种隐藏副作用会让链的复用和测试都变得困难。如果你确实需要在链里记录状态显式地把状态字段加进dict里流转比闭包变量更可控。最后再分享一个我从踩坑里总结的习惯在链的入口和出口分别加一个探针把输入输出都打印出来。这个习惯帮我省了大量排查时间——很多所谓的状态丢失其实就是我记错了入口的数据结构。把入口固定下来问题往往就清晰了一半。