012、LangChain之Chains：串联多个LLM调用实现复杂任务
从一次诡异的API超时说起
上周帮团队排查一个生产问题：一个文档摘要系统，单次调用GPT-4时响应正常，但一旦开启“多轮优化”模式，接口就频繁报504超时。翻日志发现，每次超时都卡在同一个位置——模型在生成摘要后，又调了一次LLM做“风格润色”。问题出在哪？不是模型慢，而是两次调用之间，我们手动拼接了prompt，结果把上下文撑爆了。更坑的是，第一次调用的输出里混入了特殊字符，第二次调用时直接让tokenizer崩了。
这个场景，就是典型的“需要串联多个LLM调用”但没处理好中间状态。LangChain的Chains模块，正是为了解决这类问题而生的。
Chain是什么？别想复杂了
Chain本质上就是一个“调用编排器”。它不负责模型推理，只负责把多个步骤串起来，管理输入输出、传递中间变量、处理异常。你可以把它理解成一个带状态机的管道——每个节点要么是LLM调用，要么是工具调用，要么是条件分支。
我最早接触Chain时犯过一个错误：以为Chain必须用LangChain预定义的那些，比如LLMChain、SimpleSequentialChain。后来发现，Chain的核心是BaseChain这个抽象类，你完全可以自己继承它写自定义逻辑。预定义的Chain只是帮你省事，不是限制你。
最基础的LLMChain：别被名字骗了
LLMChain这个名字容易让人误解，以为它只能做“一次LLM调用”。实际上，它是最灵活的原子单元——接收一个输入，调用一次LLM，返回一个输出。但它的价值在于，你可以把prompt模板、输出解析器、回调函数都绑定在一起。
fromlangchain.chainsimportLLMChainfromlangchain.promptsimportPromptTemplatefromlangchain.llmsimportOpenAI# 这里踩过坑：prompt模板里的变量名必须和input_dict的key一致prompt=PromptTemplate(input_variables=["product","audience"],template="为{product}写一段面向{audience}的广告文案，要求不超过50字")chain=LLMChain(llm=OpenAI(temperature=0.7),prompt=prompt,verbose=True# 调试时打开，生产环境关掉，否则日志会刷屏)# 别这样写：chain.run(product="智能手表", audience="程序员")# run方法已经废弃，用invokeresult=chain.invoke({"product":"智能手表","audience":"程序员"})print(result["text"])注意看，invoke返回的是一个字典，里面包含text字段。如果你只想要文本，可以指定output_key参数，但我不建议这么做——保持字典结构，方便后续Chain读取中间变量。
SimpleSequentialChain：最朴素的串联
当你的任务可以拆成“先做A，再做B，B的输入就是A的输出”这种线性流程时，SimpleSequentialChain是最直接的选择。
fromlangchain.chainsimportSimpleSequentialChain# 第一步：生成产品名称chain1=LLMChain(llm=OpenAI(),prompt=PromptTemplate(input_variables=["feature"],template="根据{feature}这个特性，想一个科技产品名称，只输出名称本身"))# 第二步：为这个名称写广告语chain2=LLMChain(llm=OpenAI(),prompt=PromptTemplate(input_variables=["name"],template="为产品{name}写一句朗朗上口的广告语"))# 串联起来overall_chain=SimpleSequentialChain(chains=[chain1,chain2],verbose=True)# 输入给第一个chain，输出自动传给第二个result=overall_chain.invoke("超长续航")print(result["output"])# 注意这里key是output，不是text这里有个坑：SimpleSequentialChain默认只传递一个字符串，不传递字典。如果你的chain需要多个输入变量，它就不够用了。我见过有人硬把多个变量拼成一个字符串传进去，结果prompt解析时乱成一团——别这么干，后面有更好的方案。
SequentialChain：真正的多变量串联
当每个步骤需要不同的输入变量，或者你想保留中间结果时，用SequentialChain。
fromlangchain.chainsimportSequentialChain# 第一步：分析用户评论情感chain_analysis=LLMChain(llm=OpenAI(),prompt=PromptTemplate(input_variables=["review"],template="分析以下评论的情感倾向（正面/负面/中性）：{review}"),output_key="sentiment"# 指定输出变量名，方便后续引用)# 第二步：根据情感生成回复chain_reply=LLMChain(llm=OpenAI(),prompt=PromptTemplate(input_variables=["review","sentiment"],template="用户评论：{review}\n情感分析：{sentiment}\n请生成一个合适的回复"),output_key="reply")# 串联，指定输入输出映射overall=SequentialChain(chains=[chain_analysis,chain_reply],input_variables=["review"],# 整个chain的输入output_variables=["sentiment","reply"],# 最终输出哪些变量verbose=True)result=overall.invoke({"review":"这个产品太差了，经常死机"})print(result["sentiment"])print(result["reply"])关键点：output_key决定了这个chain的输出变量名，后续chain的prompt模板里可以直接引用。这比手动解析JSON靠谱多了——我之前手动解析LLM输出的JSON，结果模型偶尔输出格式不对，直接抛异常。用output_key让LangChain帮你处理输出解析，省心。
RouterChain：根据内容动态选择路径
有时候，你需要根据输入内容决定走哪条处理链路。比如用户问技术问题就走技术回答链，问情感问题就走情感支持链。RouterChain就是干这个的。
fromlangchain.chainsimportRouterChainfromlangchain.chains.routerimportLLMRouterChainfromlangchain.chains.router.llm_routerimportRouterOutputParser# 定义两个子链tech_chain=LLMChain(llm=OpenAI(),prompt=PromptTemplate(input_variables=["question"],template="你是一个技术专家，请用专业术语回答：{question}"))emotion_chain=LLMChain(llm=OpenAI(),prompt=PromptTemplate(input_variables=["question"],template="你是一个心理咨询师，请用温暖的语言回答：{question}"))# 路由配置router_prompt=PromptTemplate(input_variables=["input"],template="""判断以下问题属于哪个类别：技术问题或情感问题 问题：{input} 只输出类别名称：技术问题 或 情感问题""")router_chain=LLMRouterChain.from_llm(llm=OpenAI(),prompt=router_prompt,output_parser=RouterOutputParser())# 创建路由链fromlangchain.chains.routerimportMultiPromptChain chain=MultiPromptChain(router_chain=router_chain,destination_chains={"技术问题":tech_chain,"情感问题":emotion_chain},default_chain=tech_chain,# 无法识别时走默认verbose=True)result=chain.invoke("我的电脑蓝屏了怎么办")print(result["text"])这个路由机制有个隐藏问题：路由判断本身也是一次LLM调用，增加了延迟和成本。如果输入类别固定，建议用规则匹配代替LLM路由。我有个项目就是先用关键词匹配，匹配不到再走LLM路由，性能提升明显。
自定义Chain：当预定义的不够用
预定义的Chain覆盖了80%的场景，但总有那20%需要你手写。比如你要在两次LLM调用之间插入一个数据清洗步骤，或者调用一个外部API。
fromlangchain.chainsimportLLMChainfromlangchain.callbacks.managerimportCallbackManagerForChainRunfromtypingimportDict,AnyclassDataCleaningChain(LLMChain):"""在LLM调用前先清洗输入数据"""def_call(self,inputs:Dict[str,Any],run_manager:CallbackManagerForChainRun=None,)->Dict[str,Any]:# 这里踩过坑：必须调用父类的_call，否则回调不触发# 先清洗数据raw_text=inputs.get("text","")cleaned=raw_text.replace("\n"," ").strip()iflen(cleaned)>1000:cleaned=cleaned[:1000]# 截断，防止token溢出# 别这样写：直接修改inputs字典，会污染原始数据# 应该创建新字典new_inputs={"text":cleaned}# 调用父类逻辑returnsuper()._call(new_inputs,run_manager)# 使用chain=DataCleaningChain(llm=OpenAI(),prompt=PromptTemplate(input_variables=["text"],template="总结以下内容：{text}"))result=chain.invoke({"text":" 很长很长的文本... "})自定义Chain的核心是重写_call方法。注意不要重写invoke或__call__，那些是框架层面的调度逻辑。_call才是你真正需要干预的业务逻辑。
调试Chain的实用技巧
Chain一旦嵌套多层，出问题很难定位。我总结几个血泪教训：
打开verbose=True：每个chain的输入输出都会打印，虽然日志多，但能快速定位哪一步出了问题。
用回调函数记录中间状态：
fromlangchain.callbacksimportStdOutCallbackHandler handler=StdOutCallbackHandler()chain.invoke(inputs,callbacks=[handler])单步测试：不要一次性把整个SequentialChain跑通。先单独测试每个子chain，确认输入输出格式正确，再串联。
注意输出解析器：如果LLM输出格式不稳定，加一个输出解析器（如
PydanticOutputParser），让解析失败时抛出明确异常，而不是默默返回乱码。
性能优化：别让Chain变成性能瓶颈
Chain本身的开销很小，但不当使用会导致性能问题：
- 减少不必要的LLM调用：能用规则判断就别用LLM。比如路由判断，先试关键词匹配。
- 缓存LLM结果：对于相同输入，用
CacheBackedLLM缓存结果，避免重复调用。 - 异步执行：如果多个Chain之间没有依赖关系，用
asyncio并发执行。LangChain支持异步invoke，但要注意线程安全。 - 控制prompt长度：每次LLM调用都会把历史prompt传进去，Chain串联时尤其要注意。我见过一个Chain嵌套了5层，每层prompt都包含前几层的输出，最后token数爆炸。
个人经验：什么时候该用Chain，什么时候不该用
Chain最适合的场景是：流程固定、步骤明确、中间变量多。比如文档处理流水线：先提取关键信息，再生成摘要，最后翻译成多语言。
但如果你只是“调一次LLM，然后处理结果”，完全没必要用Chain。直接调LLM对象，自己写几行Python处理逻辑，比套Chain更清晰。
还有一个反模式：用Chain实现状态机。有人试图用RouterChain实现复杂的对话状态流转，结果代码变得极其难调试。这种场景建议用专门的对话管理框架，或者自己写状态机。
最后，记住Chain的核心理念：它帮你管理的是“调用之间的数据流”，而不是“业务逻辑”。业务逻辑应该写在prompt模板和自定义处理函数里，Chain只负责编排。把这个边界划清楚，你的代码会好维护得多。
