LangChain Chains实战：从工作流编排到数据流设计的完整指南

1. 从“单次调用”到“工作流编排”的思维转变

如果你已经在用LangChain跑通了一些单轮对话、基础问答的Demo，大概率会遇到一个共同的瓶颈：单次调用大模型只能完成一件相对独立的事，而真实业务场景里几乎没有“一问一答”就结束的任务。

拿我前段时间给某内部团队做的客服工单分类系统来说，完整流程至少包含：意图识别 → 工单优先级判断 → 回复内容生成 → 结构化结果输出。如果拆成四次独立调用，每次都要重新传一遍上下文，中间的逻辑切换还得自己写胶水代码。但在LangChain里，这套流程可以用Chains串起来——每个环节是一个独立单元，单元之间通过确定的接口传递数据，整体像流水线一样跑完。

这就是Chains在LangChain 0.1.x时代存在的根本意义：用标准化的方式把多个步骤编排成一个整体执行单元。它解决的核心问题不是“调用模型”，而是“如何组织调用顺序、如何传递中间结果、如何在失败时兜底”。

需要先声明一点：这篇文章基于LangChain 0.1.x到0.2.x的版本线来写。LangChain在0.3之后把LCEL（LangChain 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的输出汇总起来。

举个实际的例子。我做过一个合同条款风险扫描的工作流，流程是这样的：

1. 抽取合同里的关键条款（输出：clauses）
2. 对每条条款做风险判断（输出：risk_assessment）
3. 把风险判断结果规整成最终报告（输入：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是对应的处理Chain
• default_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：对最终回复做一遍语气检查（确保不含负面情绪）
• 输出：最终客服回复文本

对应的数据流设计如下：

这个设计里最关键的一点是：分支处理子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( model="your-model-name", temperature=0.2, max_tokens=512, )

然后构造三个LLMChain：

hw_chain = LLMChain(llm=llm, prompt=hw_prompt, output_key="reply_draft") sw_chain = LLMChain(llm=llm, prompt=sw_prompt, output_key="reply_draft") bill_chain = LLMChain(llm=llm, prompt=bill_prompt, output_key="reply_draft")

再写路由用的Prompt和Chain：

router_template = """ 给定用户输入，将其分类为以下选项之一：硬件、软件、账单。 要求：只输出一个词，不要标点，不要解释。 <<用户输入>> {user_query} <<分类结果>> """ router_prompt = PromptTemplate( input_variables=["user_query"], template=router_template, ) router_chain = LLMChain(llm=llm, prompt=router_prompt, output_key="route_type")

最后用MultiPromptChain组合：

workflow = MultiPromptChain( router_chain=router_chain, destination_chains={ "硬件": hw_chain, "软件": sw_chain, "账单": bill_chain, }, default_chain=sw_chain, verbose=False, )

跑一下：

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时传memory=xxx。

一个我踩过的坑：在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 三种高频工作流模式对比

我整理了实际项目中经常出现的三类编排模式：

在实际项目里，这三种模式经常嵌套。比如客服工作流里，流水线式负责整体结构，分支路由式负责中间决策，聚合汇总式负责最后决策依据的合并。用LangChain写之前，先在纸上画出这三个层次的嵌套关系，能极大减少返工。

5.2 从报错信息定位数据流问题

我收集了几个高频报错，并给出直接从代码层面定位的思路：

报错1：KeyError: 'reply_draft'

• 原因：某子Chain的输出key并没有出现在父Chain的output_variables里，但后续子Chain又在input_variables里引用了它
• 定位顺序：检查父Chain定义里的output_variables是否包含所有需要传给下游的key → 检查子Chain的input_variables是否跟父Chain的output_variables完全一致

报错2：ValidationError: 1 validation error ... prompt

• 原因：PromptTemplate里定义的input_variables和传入的实际dict不匹配
• 定位顺序：把该子Chain单独取出来跑一次invoke，确认它需要哪些变量 → 再查上游实际传入了哪些key

报错3：模型输出为“无”或空字符串

• 情况1：max_tokens设置太短，输出被截断
• 情况2：Prompt模板里没有充分指定输出格式，模型自由发挥
• 解决：加大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 API（SequentialChain、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(llm=llm, prompt=translate_prompt) polish_chain = LLMChain(llm=llm, prompt=polish_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 | llm

LCEL的优势是：代码更短、并行组合更容易、流式输出支持更好、对中间步骤的处理靠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真正用明白。