如果你是从这个系列一路跟过来的,前几篇文章里我们已经把提示词工程、模型参数、结构化输出这几个点单独聊明白了。但大多数初学者到了这一步都会卡一下:单次调用能做好,一旦需要“先判断用户意图,再抽取关键信息,还要分情况调用不同的模板”这种多步骤任务,代码就会变得又臭又长,而且每一步的成功都依赖上一步的文本格式对齐,调试起来特别头疼。这篇文章要聊的LLMChain、顺序链、路由链,正是解决这个问题的标准做法。我会用一组可以照着改的代码示例,带你搭出一条完整的AI工作流水线。适合那些已经能调通模型接口、又想进入真实业务编排的同学;理解这套组合拳之后,你会发现多步骤AI应用的代码结构突然清爽很多。
1. 为什么AI应用需要“链”?换个角度理解模型调用
1.1 单次调用的局限:你问一次,它答一次
先举一个最简单的例子。假设你做的是一个客服工单系统,用户提交了一段话:“我三天前买的数据线,插上电脑没反应,想退货,你们的服务太差了。”
这句话里至少有三个待办事项:识别用户情绪、判断问题类型、决定后续处理流程(退货指引还是安抚)。如果只调用一次模型,你只能在提示词里塞进所有要求,让模型一次性吐出一段结构复杂的文本。下一次需求换成“如果是技术问题,还需要让用户提供设备型号”,prompt 又要全部重写。更麻烦的是,如果这个中间结果还需要给其他系统使用,比如工单系统要记录“问题类型”字段,你就要在整段输出里做正则抽取,抽不中就报错。
所以,单次模型调用本质上是一个封闭的小黑盒:输入一段文本,输出一段文本。真实业务却需要把多个模型调用串起来,并且让中间结果可控、可追踪、可复用。链就是为这个需求设计的。
1.2 流水线的类比:每个工位只做一件事
传统的工厂流水线很好理解:一个工位只拧一颗螺丝,拧完传给下一个工位。AI应用里的“链”也是同样的逻辑。每个链负责一个单一且清晰的任务,前一个链的输出文本,经过简单的整理,变成后一个链的输入。
与传统编程管道最大的不同点是:传统管道的传输格式是强类型的,比如一个结构体跟着管道走,类型不对编译器直接报错;模型输出是文本,语言模型说错了你也只能根据字符串去解析。因此,在链式编程里,“如何约定中间文本的格式”就非常重要——这也是后面输出解析器出场的原因。
可以被链化的任务非常多:摘要、翻译、改写、分类、抽取、格式化、合规检查、润色,只要是一个可以用Prompt描述的任务,都可以包成一个链。每个链内部保持简单,复杂逻辑交给链与链之间的组合,这是整套设计的灵魂。
1.3 三种链的分工:最小单元、传送带、分拣机器人
在这一篇里,我会重点聊三种最常用的链:
- LLMChain:最小工作单元,一个提示模板加一个模型再加一个可选解析器,完成一次调用。
- 顺序链(SequentialChain):按顺序执行多条链,让前者的输出进入后者的输入。
- 路由链(RouterChain):根据输入内容的不同,把任务分发到不同的后续链。
可以用一个表格来概括它们的角色:
| 链 | 类比 | 解决什么问题 | 什么时候用 |
|---|---|---|---|
| LLMChain | 一个工位 | 单次模型调用封装 | 所有链的基本构成单元 |
| SequentialChain | 传送带 | 把A的结果送给B | 任务步骤固定、单向流动 |
| RouterChain | 分拣机器人 | 按输入类型选择不同处理路径 | 输入分类明确、分支较多 |
实际项目里,几乎可以认为所有复杂场景都是这三种链的自由组合。不要被五花八门的链名称吓到,底层都是这三个思路的变体。
2. LLMChain:先把最小工位打磨好
2.1 第一段LLMChain:一个最简单的翻译链
我以当前主流的Python AI应用框架为例,写一个最小可运行的LLMChain。假设我们要做的是“把一段项目描述翻译成英文”,然后顺便给翻译结果加一个“翻译准确度评分”。
需要说明的是,各框架的API版本迭代比较快,这里展示的是比较典型的经典写法,重点看三段式的结构:模板、模型、链的组装。无论API以后怎么变,这个结构基本是稳定的。
from langchain.llms import OpenAI from langchain.prompts import PromptTemplate from langchain.chains import LLMChain # 1. 提示模板 prompt = PromptTemplate( input_variables=["source_text", "language"], template="请把下面的文本翻译成{language},只输出译文,不要解释。\n文本:{source_text}", ) # 2. 模型 llm = OpenAI(model="gpt-3.5-turbo-instruct", temperature=0) # 3. 组装链 translation_chain = LLMChain(llm=llm, prompt=prompt) result = translation_chain.run( source_text="这个项目主要解决多步骤任务的自动化编排问题", language="英文", ) print(result)执行后,result就是模型返回的翻译文本。
这里有一个新手很容易忽略的细节:PromptTemplate里用花括号包裹的变量名,必须和run()传入的参数名、以及input_variables列表里的名字完全一致。不一致的时候,框架报错信息有时只在执行阶段才出现,而且很难一眼看出来。
2.2 三件套:模板、模型、输出解析器
一个标准的 LLMChain 一般由三部分组成:
- 提示模板:定义“这个工位要做什么”。它决定了任务的边界,也决定你控制质量的抓手。
- 模型:负责实际的推理生成。不同的任务对模型要求完全不同,比如分类任务温度可以低一些,文案创作温度可以高一些,这些参数是在模型这里控制的。
- 输出解析器:把模型返回的文本转成你需要的结构化对象,也就是把“文本”变成“数据”。
很多教程把重点放在前两个上,但我个人认为,关注输出解析器才是从“玩具代码”走向“生产可用”的关键一步。举个例子,我希望模型除了翻译,还返回一个“准确度评分”,我用一个输出解析器来保证它是 JSON 格式:
import json from langchain.output_parsers import PydanticOutputParser from pydantic import BaseModel, Field class TranslationResult(BaseModel): translated_text: str = Field(description="翻译后的文本") confidence: int = Field(description="翻译准确度评分,0到100之间的整数") parser = PydanticOutputParser(pydantic_object=TranslationResult) prompt = PromptTemplate( template="请完成翻译任务,并严格按照{format_instructions}的格式输出。\n原文:{source_text}", input_variables=["source_text"], partial_variables={"format_instructions": parser.get_format_instructions()}, )这样,链的输出就不再是一段需要靠肉眼检查的字符串,而是一个自带字段的TranslationResult对象。这个思路有一个很大的好处:后续的链可以直接按照字段名取值,而不是对文本做正则匹配,这正是整个链式架构能稳定工作的前提。
2.3 为什么LLMChain是搭一切的基础
你可以把 LLMChain 理解成乐高积木里最基础的那个方块。后面的顺序链、路由链,本质上都是在“链里面再套链”或者“管理多个链的关系”。所以写好一个 LLMChain 的质量标准,不是代码能跑,而是:
- Prompt 把任务边界写得足够清晰,同一类输入不会出现两种截然不同的输出风格;
- 输出解析器让结果稳定可解析,能直接用字段;
- 单个链的定义尽量职责单一,一个链只回答“一个问题”,如果发现 Prompt 里塞了三四个指令,说明该拆成多条链了。
我先在这篇里把 LLMChain 作为基础讲透,后面的顺序链和路由链都会在这个基础上搭建。你可以拿自己手头的一个小任务去试着包一个 LLMChain,比如“输入招聘JD,输出结构化要求清单”,这一步顺手之后,后面所有内容都会非常自然。
3. 顺序链:让前一步的输出走进下一步的输入
3.1 最简单的一串一:SimpleSequentialChain
当我有了两个 LLMChain,想按顺序执行它们时,最简单的做法就是用SimpleSequentialChain。它适合“单输入单输出”的场景:前一个链的输入只有一个,输出也只有一个,自动传给下一个链。
还是用例子说话。第一步把用户的一句话总结成一条核心观点,第二步把这个观点翻译成英文口语化表达:
from langchain.chains import SimpleSequentialChain summarize_prompt = PromptTemplate( input_variables=["content"], template="请用一句话总结这段内容的核心观点:{content}", ) translate_prompt = PromptTemplate( input_variables=["summary"], template="请把这句话翻译成英文,要求自然、口语化:{summary}", ) chain_1 = LLMChain(llm=llm, prompt=summarize_prompt) chain_2 = LLMChain(llm=llm, prompt=translate_prompt) overall_chain = SimpleSequentialChain( chains=[chain_1, chain_2], verbose=True, ) result = overall_chain.run("这是一段关于AI自动化编排的长文章……")执行的时候,overall_chain会自动把链一的结果作为链二的summary变量。它的方便之处是几乎不用写连接逻辑,但也有明显局限:隐式地把“唯一输出”传给“唯一输入”。一旦某个链有多个输入,或者想要取用上上步的某个变量,它就不够用了。
3.2 显式变量绑定的 SequentialChain
真实场景里,我遇到更多的其实是“多输入多输出”的需求。比如我想让一个链提炼文章要点,再让第二个链基于提炼结果写标题,第三个链基于原始文章和标题写摘要。这时候每一步需要哪些变量,需要显式声明。
用SequentialChain的写法大概是这个样子:
from langchain.chains import SequentialChain # 链一:提炼要点 extract_prompt = PromptTemplate( input_variables=["article"], template="请从下面的文章中提炼3个核心要点:\n{article}", ) # 链二:基于原始文章和要点写标题 title_prompt = PromptTemplate( input_variables=["article", "key_points"], template="根据这篇文章的要点,写一个吸引人的标题。\n要点:{key_points}\n文章:{article}\n标题:", ) # 链三:基于要点和标题生成摘要 summary_prompt = PromptTemplate( input_variables=["key_points", "title"], template="请用100字以内写一个摘要,要涵盖以下要点,同时呼应标题。\n要点:{key_points}\n标题:{title}\n摘要:", ) chain_1 = LLMChain(llm=llm, prompt=extract_prompt, output_key="key_points") chain_2 = LLMChain(llm=llm, prompt=title_prompt, output_key="title") chain_3 = LLMChain(llm=llm, prompt=summary_prompt, output_key="summary") workflow = SequentialChain( chains=[chain_1, chain_2, chain_3], input_variables=["article"], output_variables=["key_points", "title", "summary"], ) result = workflow({"article": "...某篇文档..."})这段代码里,最核心的设计是output_key。每一个链的输出都会以一个 key 存入一个“中间变量池”,后一个链可以从池子里取它需要的变量。这就让步骤间的调用不再是隐式的“唯一输出传给唯一输入”,而是完全可控的变量读取。你可以看到链一输出了key_points,链二同时拿article和key_points作为输入,链三则只需要key_points和title,从池子里各取所需。
3.3 中间结果可见性:排查问题时的关键
顺序链的另一个隐藏价值,是让“中间结果”变得可见。
如果不使用链,两个模型调用之间的文本传递发生在你的业务代码里,一旦第二步结果不好,你很难判断是第二步 Prompt 的问题,还是第一步的输出就不对。用链以后,你可以给整个SequentialChain或每个子链打开详细的执行日志(很多框架里是verbose=True或监听回调),把每一步的输入输出都打印出来,然后逐段检查。
我自己的经验是:顺序链的调试顺序永远是“倒序逐环节检查”而不是“从头开始”。先看最终输出哪里不对,再倒回到它的直接输入,判断问题是来自这一步本身还是上游传进来的数据。
讲个真实教训:我之前写过一个内容运营流水线,第一链负责“提取文章关键词”,第二链负责“生成标题”。标题质量一直很差,我反复调第二链的提示词,没效果。后来把第一链的输出打印出来才发现,第一链把关键词列表和文章放在同一个文本区域,第二链被大量原文干扰,抓不住重点。排除了上游问题之后,标题质量一下子就好了。这个问题单看第二链是不可能发现的,这就是链编排带来的“结构化调试”优势。
4. 路由链:在流水线上装一个分拣机器人
4.1 路由的本质:先分类、再分发
顺序链解决的问题是“流程固定”。但很多业务并非固定一条路走到底。比如,一个客服系统面对“退货”“技术问题”“投诉”“订单查询”四种输入,最优的处理提示词完全不同——退货要讲流程,技术问题要根据型号排查,投诉要先道歉安抚。如果强行把所有要求塞进一个 Prompt,模型的选择和语气很容易互相干扰。
路由链的标准做法,是先让模型对输入做一次分类,然后根据分类结果,把输入文本转交给对应的目标链。它就像快递分拣中心:包裹到了,先看地址属于哪个片区,再送上对应的卡车。这个“分类”动作本身也是一次模型调用,因为它要理解自然语言,判断输入的情感倾向和意图类型。
4.2 组成结构与关键配置
一条完整的路由链,通常由四部分组成:
- 目的地列表:一组“名称 + 描述”。描述的作用是告诉路由模型这个分支适合处理什么类型的问题,描述写得太笼统,模型会分错。
- 路由提示模板:框架负责把目的地列表格式化成一个选择题,让模型选择其中一个名字。
- 路由链:专门负责输出“目的地名字 + 原始输入”的链。它会用输出解析器把模型的回答转成结构化的路由决策。
- 目标链列表 + 默认链:根据路由结果,把任务交给不同的后续 LLMChain;当路由模型选择不在列表里或者解析失败时,交给默认链处理,避免整个流程直接崩溃。
这里的目的地列表,质量直接影响整个路由的准确率。描述必须写清楚“这个分支适合什么输入,不适合什么输入”,最好再给一两个典型例子。
4.3 一个客服工单分拣的完整示例
我写一个最小可运行的客服工单路由示例。这里的目标链分别是“技术咨询”“投诉处理”“订单问题”。先定义三个目标链:
from langchain.chains.router import LLMRouterChain from langchain.chains.router.multi_prompt_prompt import MULTI_PROMPT_ROUTER_TEMPLATE from langchain.chains import MultiPromptChain tech_prompt = PromptTemplate( input_variables=["input"], template="你是一名技术工程师。请用通俗易懂的语言解答这个技术问题,并给出排查步骤:\n{input}", ) complaint_prompt = PromptTemplate( input_variables=["input"], template="你是一名客服经理。请先向用户表达歉意,再提出可行的补偿或解决方案:\n{input}", ) order_prompt = PromptTemplate( input_variables=["input"], template="你是一名订单处理专员。请帮用户梳理订单状态并给出后续操作指引:\n{input}", ) destination_chains = { "tech": LLMChain(llm=llm, prompt=tech_prompt), "complaint": LLMChain(llm=llm, prompt=complaint_prompt), "order": LLMChain(llm=llm, prompt=order_prompt), }接下来定义目的地描述。描述写得好不好,直接决定模型分到哪个分支:
destinations = [ "tech:处理编程、设备配置、软件报错、API接入等技术类问题。", "complaint:处理用户投诉、不满情绪、赔偿诉求、服务态度问题。", "order:处理订单查询、退换货、物流信息、发票等交易相关问题。", ]然后把目的地列表交给路由模板,由路由链输出决策。使用框架自带的多提示路由模板,可以省去自己拼接目的地格式的麻烦:
router_template = MULTI_PROMPT_ROUTER_TEMPLATE.format(destinations="\n".join(destinations)) router_prompt = PromptTemplate( template=router_template, input_variables=["input"], output_parser=RouterOutputParser(), ) router_chain = LLMRouterChain(llm=llm, prompt=router_prompt) multi_chain = MultiPromptChain( router_chain=router_chain, destination_chains=destination_chains, default_chain=default_chain, )执行时,只要调用:
response = multi_chain.run("我买的数据线三天没到货,帮忙查一下物流")路由链会先识别“这是订单问题”,然后把输入交给order链,得到的结果就是订单专员风格的回复。
这里的default_chain是兜底链,可以是任意一个 LLMChain。比如统一使用一个“通用客服”链来回答那些分类不清的问题,保证系统在模型判断失误时依然有回应。
需要特别提醒的是,destination_chains字典的键名,必须和目的地列表里的名字保持一致,比如这里都是tech、complaint、order。否则路由选到了tech,却去字典里找technique,整个流程会直接抛异常。
4.4 兜底链:模型选错分支的保底措施
路由链最容易出问题的点,是模型分类不准。比如用户说“你们客服根本不解决问题”,语气明显不满,但核心诉求可能是订单问题,路由模型可能归到“投诉”而不是“订单”,或者落在未知分支上。
我的建议是永远配置一个默认链,同时把目的地描述写得更精确。另外,如果预算允许,可以让路由模型使用更擅长指令遵循的模型,并用较低的温度,让分类结果更可控。还有一个不显眼但很重要的点:RouterOutputParser的解析规则依赖模型输出格式,模型偶尔会冒出一段额外解释,导致解析失败,这时默认链的存在能保证整个系统不崩溃。
不过需要提醒一句:路由链里“分类”和“业务处理”都调用了模型,成本是两次调用。如果确实想省钱,也可以只让路由模型返回一个枚举ID,用更小的模型来完成分类,主模型只负责业务生成。
5. 三段组合实战:把投稿变成一篇排版好的成稿
5.1 业务场景:从用户投稿到多渠道成稿
把前面三招合在一起,我们做一个接近真实业务的Demo。假设一个小型内容社区收到一篇用户投稿,自动处理流程要完成四件事:
- 提炼投稿的核心观点(LLMChain);
- 判断内容属于“技术分享”“生活记录”“观点评论”中的哪一类(路由链);
- 根据分类,用不同的风格扩写成一篇有感染力的成稿(三个目标链);
- 根据成稿和观点,生成一条社交媒体的发布摘要(顺序链收尾)。
这个流程中,顺序链保证步骤1到步骤4单向推进,路由链控制步骤3走哪条路线。可以看到,顺序链和路由链并不是互斥的,而是嵌套使用的。
5.2 组合结构与代码骨架
组合的目的不是炫技,而是让每段Prompt保持单一职责:第一步只管提炼观点,不管风格;第二步只管分类,不管措辞;第三步按类型扩写,不混风格;第四步只写摘要,不管标题。代码骨架如下:
# 第一步:提炼观点 extract_chain = LLMChain( llm=llm, prompt=PromptTemplate( input_variables=["raw_text"], template="提炼下面投稿的核心观点,用3个关键词加1句话概括:\n{raw_text}", ), output_key="core_view", ) # 第三步的三个扩写目标链 tech_write_chain = LLMChain( llm=llm, prompt=PromptTemplate( input_variables=["core_view", "raw_text"], template="你是技术博客编辑,请把这篇投稿扩写成技术分享风格,突出实操步骤:\n核心观点:{core_view}\n原稿:{raw_text}", ), output_key="article", ) life_write_chain = LLMChain( llm=llm, prompt=PromptTemplate( input_variables=["core_view", "raw_text"], template="你是生活博主,请把这篇投稿扩写成轻松亲切的生活记录风格:\n核心观点:{core_view}\n原稿:{raw_text}", ), output_key="article", ) opinion_write_chain = LLMChain( llm=llm, prompt=PromptTemplate( input_variables=["core_view", "raw_text"], template="你是观点专栏作者,请把这篇投稿扩写成有立场、有论据的观点评论:\n核心观点:{core_view}\n原稿:{raw_text}", ), output_key="article", ) # 第四步:生成摘要 summary_chain = LLMChain( llm=llm, prompt=PromptTemplate( input_variables=["core_view", "article"], template="根据核心观点和成稿,生成一条不超过50字的社交媒体摘要:\n核心观点:{core_view}\n成稿:{article}\n摘要:", ), output_key="summary", )注意上面三个目标链的输出变量都叫article,这是故意的,因为后续摘要链只关心“最终成稿”,不需要区分它来自哪个分支。但如果后面还需要保留“是哪条分支生成的”这类信息,就应该让不同链输出不同的 key。
路由链部分和第4章类似,把destination_chains替换成上述三个扩写链即可:
destination_chains = { "tech": tech_write_chain, "life": life_write_chain, "opinion": opinion_write_chain, } router_chain = LLMRouterChain(llm=router_llm, prompt=router_prompt) route_step = MultiPromptChain( router_chain=router_chain, destination_chains=destination_chains, default_chain=tech_write_chain, # 默认走技术风格,保证兜底可用 )然后组装顺序链。这里要特别留意:SequentialChain的input_variables和output_variables必须覆盖所有用到的变量。比如第2步路由需要的是core_view或raw_text,第3步扩写需要raw_text,第4步摘要需要core_view和article。把所有链的输出 key 整理成一张清单,再填进SequentialChain:
pipeline = SequentialChain( chains=[extract_chain, route_step, summary_chain], input_variables=["raw_text"], output_variables=["core_view", "article", "summary"], ) result = pipeline({"raw_text": "今天分享一个我自己写的脚本,它能自动备份照片..."})这段代码能跑通的前提是route_step这个MultiPromptChain能作为SequentialChain的一个步骤被调用。实际项目里,这一步可能需要包一层自定义适配器。但即使 API 细节因框架版本而异,组合思路是不变的:一个大链的每个节点,本身可以是一个子链,这在框架设计上是完全允许的。
5.3 执行过程与变量追踪
整个流水线执行时,我想强调一个跟踪技巧:别等最终结果不对劲才开始打日志,一开始就在组合层开启详细输出。把每个链的 prompt 输入和模型输出打印出来,你就相当于给流水线装了一排观察窗。哪个环节输出格式不对、哪个环节跑错了分支,立刻就能定位。
上面这个Demo虽然简单,但已经涵盖了三种链的所有组合方式:普通步骤用 LLMChain、固定流程用 SequentialChain、分支选择用 RouterChain。换一个业务场景,比如“简历筛选+面试题生成+邮件通知”,也只是把节点内容换掉,骨架完全一样。
6. 我在真实项目中踩过的坑与调整思路
6.1 变量命名是隐性的接口协议
前面提到过,链系统里靠 key 来传递变量。链一多了,变量名就成了团队里隐性约定。我见过最惨的一次事故,是两条链都把输出 key 写成了result,后一条链覆盖前一条,最终结果在两种数据之间随机抖动,排查了很久。
我的做法是给中间变量加“步骤前缀”:step1_core_view、step2_category、step3_article。这样即使链很多,一眼就能看出变量来自哪个环节。输入变量用统一名字input,只在入口处从业务数据转一次,避免同一含义出现多个叫法。
6.2 模型输出不稳定时必须留后路
模型输出是概率性的,路由、解析、分类这些环节,都可能因为一个标点符号格式不对而失败。任何依赖模型输出的关键环节,都要有一个失败之后的保底方案。
我在生产里做的三个保底是:
- 路由链一定配默认链,并且默认链的业务回复质量也要达标,不能成为明显降级;
- 输出解析失败时做一次“纠错重试”,把模型的原始回答和报错信息一起塞回提示词,让模型重新输出合法结果;
- 每个链的关键节点做超时和异常捕获,保证哪怕某一步失败,也能返回给调用方一个清晰的状态码和中间产物,而不是一个莫名其妙的空值。
这三条不是锦上添花,而是链式应用稳定性的基本盘。
6.3 别为了用链而用链:拆链的收益要算清楚
很多人学完链以后,容易走向另一个极端:把所有任务都拆成七八条链,觉得这样“很架构”。但每条链就是一次模型调用,带来的是时间和钱的双重开销。两个步骤如果永远是一起执行、不需要中间干预,那就把它们放在同一个 Prompt 里合并成一个链。
我个人的判断标准很简单:只有当中间结果要被复用、需要被校验、或者需要走不同分支时,才值得拆链。如果只是为了“让提示词短一点”而拆,收益很小。拆链就像拆函数,不是为了好看,而是为了复用和隔离复杂度。
6.4 用“执行台账”快速定位问题
最后分享一个我一直在用的调试习惯:给链组加一个记录执行过程的回调,把每一步的“链名、输入变量、输出变量、耗时、token数”写进一个结构化日志。排查问题时,我习惯先按住Ctrl+F搜关键词,找到哪条链的输出为空或格式不对,再针对那一条链去细查提示词。
这个台账相当于给流水线装了一套仪表盘。项目上线之后,它也能帮你做成本统计:哪条链消耗token最多、平均耗时最长,这样你优化性能的时候就有数据支撑,而不是靠猜。
到这里,LLMChain、顺序链、路由链的组合打法就讲完了。说实话,链这个抽象概念刚出现的时候我也有点抗拒,觉得“不就是包了一层吗”。真正在项目里大规模使用以后,我才意识到它解决的不是“怎么写代码”的问题,而是“怎么让多个模型调用像一个系统一样协作”的问题。你现在如果手头有一个多步骤的AI需求,建议先别写代码,拿张纸画一下:哪些步骤是固定的,哪些步骤需要分叉,哪些结果要留到后面用。把图画清楚,再来套这三种链,会顺畅很多。