1. 从“单次调用”到“工作流编排”的思维转变如果你已经在用LangChain跑通了一些单轮对话、基础问答的Demo大概率会遇到一个共同的瓶颈单次调用大模型只能完成一件相对独立的事而真实业务场景里几乎没有“一问一答”就结束的任务。拿我前段时间给某内部团队做的客服工单分类系统来说完整流程至少包含意图识别 → 工单优先级判断 → 回复内容生成 → 结构化结果输出。如果拆成四次独立调用每次都要重新传一遍上下文中间的逻辑切换还得自己写胶水代码。但在LangChain里这套流程可以用Chains串起来——每个环节是一个独立单元单元之间通过确定的接口传递数据整体像流水线一样跑完。这就是Chains在LangChain 0.1.x时代存在的根本意义用标准化的方式把多个步骤编排成一个整体执行单元。它解决的核心问题不是“调用模型”而是“如何组织调用顺序、如何传递中间结果、如何在失败时兜底”。需要先声明一点这篇文章基于LangChain 0.1.x到0.2.x的版本线来写。LangChain在0.3之后把LCELLangChain Expression Language提到了绝对核心的位置但Chains的编排思想、数据流设计、以及与之配套的Memory、OutputParser、Callback等模块交互方式依然是理解新版LCEL的最佳跳板。而且在实际维护的老项目里SequentialChain、RouterChain这类结构仍然大量存在。2. 拆解Chains的核心类型与设计意图2.1 LLMChain最基础但最容易被轻视的环节LLMChain是整个Chains体系的地基。一句话概括它把一个Prompt模板、一个LLM实例、一个输出解析器打包成了一个可复用单元。很多初学者做了两个Demo就开始直接上手SequentialChain结果回头排查问题的时候发现最简单的那一环就埋了雷。比如Prompt模板里定义的变量名和Chain输入字典里的key对不上这种错误在LLMChain里会直接抛出KeyError但当你把它嵌套进复杂Chain里时报错信息会被吞掉变成莫名其妙的输出。我建议每一个要上Chains的工程起步阶段都明确把LLMChain单独写一层。它带来的好处是Prompt模板可以独立做版本管理不跟业务逻辑耦合LLM实例模型名、温度、max_tokens可以在一个地方统一配置如果后续要替换模型只需要改一个LLMChain的定义下游全部自动适配实际项目里单轮LLMChain最常见的落点是“格式化输出”。比如让模型输出JSON同时又不想每次都在Prompt里重复写“你是一个JSON生成器”就可以抽成一个模板固定的LLMChain。2.2 SequentialChain串行执行时的数据流设计当业务逻辑是“先做A再做B最后做C”这种严格顺序时SequentialChain就是最直观的编排工具。它有两个变体SimpleSequentialChain单输入单输出和 SequentialChain多输入多输出。SimpleSequentialChain的写法很友好输出直接作为下一个Chain的输入适合“扩写→翻译→润色”这类一条线走到黑的场景。但真实业务里大部分流程是“多输入多输出”的。这时候你用SequentialChain核心要理解它的数据传递规则每个子Chain有一个input_variables和output_variables父Chain通过input_variables指定要传给子Chain哪些key同时通过output_variables把所有子Chain的输出汇总起来。举个实际的例子。我做过一个合同条款风险扫描的工作流流程是这样的抽取合同里的关键条款输出clauses对每条条款做风险判断输出risk_assessment把风险判断结果规整成最终报告输入risk_assessment 原始合同片段如果用SimpleSequentialChain第3步就拿不到原始合同片段了因为你只有一条数据链前一步的输出就是后一步的全部输入。用SequentialChain的话第3步可以同时接收第2步的输出和原始输入数据这样生成报告时才能把风险点定位到具体的合同条款上。数据流设计是SequentialChain的灵魂。我个人的经验是先把整个流程里需要传递的数据列成一张表标清楚每个步骤的输入key和输出key再动手写代码。直接凭感觉写大概率会在连续传递三个以上Chain时出现key穿透或者key覆盖的问题。2.3 RouterChain分支选择的工程化方案RouterChain的出现是为了解决“按条件选择不同处理路径”的问题。它的本质是先用一个LLM调用判断当前输入应该走哪条分支然后根据判断结果把输入路由到对应的Chain上。这里有一个很容易踩的坑RouterChain的判断是基于LLM的不是硬编码的规则。所以它天生带有不确定性。如果你要做的事是可以明确写成if-else的比如“当type字段等于A时走A分支”那完全没必要用RouterChain直接业务代码判断更稳定。RouterChain真正适合的场景是分支条件无法通过结构化字段判断必须靠语义理解。比如客服系统的工单类型判断用户不会提交一个type字段你需要让模型从一句“我的电脑蓝屏了但是耳机还能用”里判断出这是硬件报修还是配件咨询。实现方式上RouterChain的主要组件包括PromptTemplate决定以什么格式要求模型输出“应该走哪条路由”LLMChain负责执行这个判断destination_chains一个字典key是路由名value是对应的处理Chaindefault_chain模型无法判断时兜底用的RouterChain用起来有点讲究。输出的路由名称必须和destination_chains里的key完全一致否则会静默走default_chain。我在项目里遇到过一个问题模型在输出路由名时加了个引号导致匹配失败排查了很久才发现是输出格式问题。解决办法是在路由Prompt里用few-shot示例把输出格式限定死。2.4 并行与TransformChain补充说明除串行和路由外还有一个比较容易被忽略的ParallelChain在LangChain 0.2某些版本里称为Parallel。它适合的场景是多个互不依赖的步骤可以同时执行。但说实话ParallelChain在高版本LangChain里逐渐被LCEL的并行机制取代了。如果你想在0.3.x里做并行直接用LCEL把多个Runnable用字典组合起来就行写法更简洁。另外TransformChain是一个轻量的工具型Chain它不调用LLM而是通过自定义函数对数据进行预处理或后处理。它非常适合放在SequentialChain的最前面做数据清洗或者放在最后面做结果格式化。比如模型输出的内容里有markdown标记你想剥掉再发到下游系统就可以写一个TransformChain。3. 手把手实现一个客服工单编排工作流3.1 场景设定与数据流规划为了把前面几种Chain串起来我构造一个相对完整的场景客服工单自动处理工作流。需求描述输入一段用户问题文本步骤1判断工单类型硬件/软件/账单/其他步骤2a如果是硬件问题生成维修建议步骤2b如果是软件问题生成重启/重装指引步骤2c如果是账单问题生成账单解释步骤3对最终回复做一遍语气检查确保不含负面情绪输出最终客服回复文本对应的数据流设计如下步骤输入key输出key说明路由判断user_queryroute_type决定下游分支分支处理三选一user_query route_typereply_draft不同分支不同Prompt语气检查reply_draftfinal_reply统一后处理这个设计里最关键的一点是分支处理子Chain必须同时接收route_type作为输入。因为我们要在分支Chain的Prompt模板里明确告诉模型“你现在的角色是硬件维修工程师”而这个角色信息就是route_type决定的。3.2 逐步配置三个核心Chain下面我给出可直接运行的代码。注意版本依赖需要使用langchain 0.1.x版本。from langchain.chains import LLMChain, RouterChain from langchain.chains.router import MultiPromptChain from langchain.llms import OpenAI from langchain.prompts import PromptTemplate先定义所有子Chain的Prompt模板。这里我刻意让不同分支的Prompt结构保持一致方便后续替换模型或调整prompt时维护# 硬件分支 hw_prompt PromptTemplate( input_variables[user_query], template你是一名硬件维修工程师。针对用户问题{user_query}给出简洁的维修建议。 ) # 软件分支 sw_prompt PromptTemplate( input_variables[user_query], template你是一名软件支持工程师。针对用户问题{user_query}给出软件修复指引。 ) # 账单分支 bill_prompt PromptTemplate( input_variables[user_query], template你是一名账单客服。针对用户问题{user_query}解释可能的账单原因。 )接着初始化一个统一的LLM实例这里我用一个占位的模型对象你可以把它替换成你实际使用的模型不管是什么提供商只要实现了LangChain的LLM接口就行llm ChatOpenAI( modelyour-model-name, temperature0.2, max_tokens512, )然后构造三个LLMChainhw_chain LLMChain(llmllm, prompthw_prompt, output_keyreply_draft) sw_chain LLMChain(llmllm, promptsw_prompt, output_keyreply_draft) bill_chain LLMChain(llmllm, promptbill_prompt, output_keyreply_draft)再写路由用的Prompt和Chainrouter_template 给定用户输入将其分类为以下选项之一硬件、软件、账单。 要求只输出一个词不要标点不要解释。 用户输入 {user_query} 分类结果 router_prompt PromptTemplate( input_variables[user_query], templaterouter_template, ) router_chain LLMChain(llmllm, promptrouter_prompt, output_keyroute_type)最后用MultiPromptChain组合workflow MultiPromptChain( router_chainrouter_chain, destination_chains{ 硬件: hw_chain, 软件: sw_chain, 账单: bill_chain, }, default_chainsw_chain, verboseFalse, )跑一下result workflow.invoke({user_query: 我的电脑开机黑屏风扇在转但是屏幕不亮}) print(result[text])输出结果会是类似“针对用户问题我的电脑开机黑屏...建议先检查内存条是否松动重新插拔后尝试开机...”这样一条回复。3.3 完整流程整合与输入输出对齐上面的代码能跑通但离“可用”还有一段距离。原因在于MultiPromptChain的输出结构比较“裸”它直接返回最终子Chain的text中间的路由判断结果、中间产物全都丢了。对于调试和日志追踪来说这是不可接受的。所以我建议在实际工程里做一层封装把输入输出对齐清晰化。def process_ticket(user_query: str) - dict: # 第一步路由决策 route_result router_chain.invoke({user_query: user_query}) route_type route_result[text].strip() # 第二步选择分支 chain_map { 硬件: hw_chain, 软件: sw_chain, 账单: bill_chain, } target chain_map.get(route_type, sw_chain) # 第三步执行分支 reply target.invoke({user_query: user_query}) return { route_type: route_type, reply: reply[text], query: user_query, }这样封装之后你拿到的结构化结果可以直接写入工单系统也可以对接后续的语言检查步骤。而且调试时只要打印process_ticket的返回值就能看到完整的数据流。我个人强烈建议不要把多条逻辑都塞进一个MultiPromptChain的invoke里。它看起来很简洁但出了问题你只能看到一个text字段根本不知道模型是怎么路由的。手动拆步骤每一步都打印才是能持续维护的写法。4. 深入聊聊Chain背后的数据流与状态管理4.1 为什么Chain的输入输出是“字符串”这个问题很致命在装过几个项目之后我越来越觉得Chain的输入输出规范是整个模块里最需要提前设计的地方。默认情况下LLMChain的输出是一个带text字段的LLMResult对象你通过output_key把它变成chain_output里的一个key。但很多人忽略的是不同的Chain对输入类型的容忍度不一样。例如LLMChain对输入的处理就有一个隐式规则如果你传入一个字符串它会把它当作input变量的值如果你传入一个dict它必须包含Prompt模板里定义的所有input_variables。这两种行为在单Chain调用时看起来都正常一旦进了SequentialChain父Chain传给子Chain的永远是dict这时候如果你的子Chain模板里的变量名跟父Chain给出的key对不上就直接报错。所以我的建议是在你设计第一个Chain之前先花十分钟把整个流程的输入输出key全部定下来写成一个字典结构然后再去写代码。不要边写边定义key后面会失控。4.2 Memory机制在Chain里的正确使用方式当你的工作流需要多轮对话时比如用户先问“我的电脑蓝屏了”然后追问“怎么解决”这时候你必须把历史对话塞进Chain的上下文里。LangChain里对应的组件是ConversationBufferMemory或更高级的ConversationSummaryMemory。Memory与Chain结合时有一个隐含约束Memory进入Chain的时机是模板渲染阶段而不是调用LLM阶段。也就是说如果你的Prompt模板里定义了history变量那么你要么在传入Chain的input dict里显式给history赋值要么在构造Chain时传memoryxxx。一个我踩过的坑在SequentialChain里Memory不能直接挂在父Chain上传递给子Chain。子Chain如果想读历史记录必须在子Chain自己的构造函数里带memory参数。换句话说多Chain场景下Memory是分散的每个Chain只能看到自己挂载的那一段历史。如果你想让整个流程共享一段对话历史最稳妥的办法是给所有子Chain都挂同一个memory实例有些版本可能不支持默认行为建议看版本docstring。4.3 输出解析器在这里起什么作用严格来说OutputParser不属于Chain本身但Chain的最终效果很大程度上取决于你如何解析模型的输出。以RouterChain为例如果你不对模型输出做严格的格式封装它就可能输出一段“用户说的情况更像是硬件问题因此我选择硬件分支”这样的自然语言导致后续匹配失败。我在实战里总结了一套行之有效的做法第一步要求模型只输出一个JSON对象包含route_type和reason两个字段。第二步用StructuredOutputParser设定Pydantic模型做自动解析。第三步把解析成功与否作为重试信号解析失败就重新调用路由Chain最多重试两次。这套逻辑看起来简单但能帮你规避掉很多“模型输出不规范导致工作流断裂”的隐性故障。LangChain自带OutputFixingParser可以在解析失败时自动让LLM修复输出格式但用下来感觉它在复杂嵌套结构上不太稳定所以我在关键路径上宁可自己写重试逻辑。5. 常用编排模式与常见问题排查实录5.1 三种高频工作流模式对比我整理了实际项目中经常出现的三类编排模式模式适用场景核心Chain注意点流水线式数据清洗→处理→输出报告SequentialChain关注所有中间输出key的传递分支路由式语义分类→不同处理策略RouterChain 多个LLMChain路由输出格式必须严格聚合汇总式多维度分析同一份文本并行Chain / 多个LLMChain合并注意合并顺序避免上下文丢失在实际项目里这三种模式经常嵌套。比如客服工作流里流水线式负责整体结构分支路由式负责中间决策聚合汇总式负责最后决策依据的合并。用LangChain写之前先在纸上画出这三个层次的嵌套关系能极大减少返工。5.2 从报错信息定位数据流问题我收集了几个高频报错并给出直接从代码层面定位的思路报错1KeyError: reply_draft原因某子Chain的输出key并没有出现在父Chain的output_variables里但后续子Chain又在input_variables里引用了它定位顺序检查父Chain定义里的output_variables是否包含所有需要传给下游的key → 检查子Chain的input_variables是否跟父Chain的output_variables完全一致报错2ValidationError: 1 validation error ... prompt原因PromptTemplate里定义的input_variables和传入的实际dict不匹配定位顺序把该子Chain单独取出来跑一次invoke确认它需要哪些变量 → 再查上游实际传入了哪些key报错3模型输出为“无”或空字符串情况1max_tokens设置太短输出被截断情况2Prompt模板里没有充分指定输出格式模型自由发挥解决加大max_tokens同时在模板结尾给出few-shot示例5.3 排查技巧给Chain加日志与回调跟我之前说的一样把链条拆开每一步都打印中间结果。LangChain提供CallbackHandler机制可以在LLM开始请求前、结束请求后、chain开始前、chain结束后等节点注入日志。我用的是极简方案自己定义一个回调类重写on_chain_start、on_chain_end、on_llm_end几个方法把关键数据塞进一个全局Logger里。这样跑一次工作流就能看到完整调用链router_chain → hw_chain → final。排查效率和直接看控制台日志完全不是一个量级。给还没这么干过的读者一个具体参考from langchain.callbacks.base import BaseCallbackHandler class ChainLogger(BaseCallbackHandler): def on_chain_start(self, serialized, inputs, **kwargs): print(f[CHAIN START] {serialized.get(name, unknown)}) print(f inputs: {inputs}) def on_chain_end(self, outputs, **kwargs): print(f[CHAIN END] outputs keys: {list(outputs.keys())}) print(f outputs: {outputs})把logger实例传到Chain构造函数的callbacks参数里或者在invoke的时候作为config传进去都能看到效果。注意0.1.x和0.2.x的调用方式略有差异0.2.x里更推荐在invoke时传config{callbacks: [logger]}。6. 新旧范式对比Chain API与LCEL的取舍6.1 Chain API的不可替代之处在新版本LangChain里Chain API的很多能力被LCEL表达式替代了。你必须清楚两者各自的优劣势才能在项目选型时做决策。Chain APISequentialChain、RouterChain等的核心优势是概念直观、梯级明确、对新手友好。每个节点是一个“有名有姓”的对象你可以给它命名、挂回调、做单元测试。这在团队协作时非常关键因为代码读起来是有结构的。另外一个实际优势是Chain API在0.1.x时代非常成熟各种第三方集成比如向量存储、Agent都围绕它做了适配。如果你的基础设施还停留在0.1.x那用Chain API是性价比最高的选择。6.2 LCEL的核心优势与迁移建议LCEL是在0.2.x时代正式推向主推位置的它的核心概念是“万物皆Runnable”并通过管道符|对Runnable进行任意组合。管道符左边输出的东西会自动成为右边Runnable的输入这使得代码极其简洁。举一个对比。用Chain API实现“翻译并润色”translate_chain LLMChain(llmllm, prompttranslate_prompt) polish_chain LLMChain(llmllm, promptpolish_prompt) sequential SequentialChain( chains[translate_chain, polish_chain], input_variables[text], output_variables[polished_text], )用LCEL实现同样逻辑from langchain_core.runnables import RunnableLambda pipeline translate_prompt | llm | polish_prompt | llmLCEL的优势是代码更短、并行组合更容易、流式输出支持更好、对中间步骤的处理靠RunnablePassthrough等工具实现灵活度极高。缺点是数据流不透明一旦链路报错定位难度较大。同类链路简单时LCEL效率很高复杂链路时建议混合使用。如果你的项目是从零开始的并且主要做数据流水线类任务我建议直接用LCEL。但如果你在维护一个已经用Chain API写好的工单系统没必要为了“新”而全部重写。先把老的稳定链路留着新功能模块逐步用LCEL替换等验证稳定后再统一收口。7. 基于经验的工作流优化建议最后把这几年实际编排Chain的一些体会汇总一下这些都是踩过坑后才明白的。第一不要追求“一个Chain干所有事”。有人喜欢把路由判断、分支处理、最终润色全部塞进一个大Chain的Prompt模板里让模型“自由发挥”。短看方便长看灾难。因为你没法对中间步骤做局部的替换和调试。拆成最小粒度的Chain每次只让模型做一件事反而更稳。第二一定要给每个子Chain一个清晰的名字。命名规则我一般用“动词对象”例如extract_clauses、assess_risk、polish_reply。名字清晰的最大价值在于日志追踪时能一眼定位链路位置尤其当你的路由分支达到五个以上时。第三温度参数要根据子Chain的角色差异化设置。路由判断用低温0到0.2确保分类稳定文案生成用中温0.3到0.5显得自然语气调整用高温0.5到0.7也未尝不可。很多人一套参数走天下结果路由判断经常“自由发挥”这就是原因之一。第四要对最终输出做一次兜底校验。LLM输出什么我们都控制不了所以一定要在链路末尾加一层校验。哪怕只是检查输出非空、长度在合理区间、没有异常字符也能拦截掉大量低级错误。更严谨的做法是定义Pydantic模型做结构校验像我之前分享的客服工单例子final_reply必须是非空字符串roate_type必须在白名单里否则就抛异常重试一次。第五关于性能。Chain API的调用模式是同步串行如果某个子Chain调用耗时2秒整条链就是累加耗时。如果分支之间互不依赖尽量考虑并行如果受限于接口限流可以考虑把非关键步骤做成异步回调或者用缓存机制对相同输入做幂等跳过。第六也是最后一点一定要在项目初期就考虑好“如何优雅地跳过某个Chain”。真实场景里经常出现“用户只问了一个简单问题不需要走完整流程”。你可以用一个前置的轻量判断甚至不用LLM比如关键词匹配如果命中简单问题就直接走single_chain不需要进入的复杂流程这样整体延迟和费用都能显著下降。这个逻辑听着很简单但很多项目一开始没做后面加的时候就不得不动主干结构了。我个人在实际操作中体会最深的是Chains的价值不在于它提供了多少“开箱即用”的工具而在于它强迫你以“流程”的眼光去审视任务。你画清楚数据流向拆清楚子任务边界剩下的就是拼接工作。框架总在迭代但“拆解、编排、兜底”这套方法论不太会过时。希望这篇基于项目的拆解能帮你把LangChain的Chains真正用明白。
