链式调用实战用LangChain构建多步骤复杂任务流水线上一篇把LangChain五个核心概念过了一遍最后提到要用LCEL写链式调用。这篇就来动手从最基础的链开始一步步搭到完整的多步骤任务流水线。先回答一个基本问题。为什么我们需要链。单次调用不够用你写第一个AI应用的时候大概率是这样干的。拼一段提示词调一次模型拿结果返回。一个来回搞定简单直接。但真实业务很少这么省心。用户让你写一篇技术博客你得先生成大纲大纲确认后再逐段扩写最后还要润色。用户问你一个专业问题你可能要先判断问题属于哪个领域再走对应的检索流程最后组织回答。这些场景一个模型调用根本搞不定。我自己踩过一个坑。做一个文档摘要功能一开始想偷懒一个提示词让模型又提炼要点又润色又排版。结果模型顾此失彼要点抓不全文字也不够通顺。拆成三步之后先提炼再润色最后排版每一步只干一件事质量立马好了一截。这就是链式调用的价值。把一个复杂任务拆成几个小步骤每个步骤用一个链处理前一步的输出喂给下一步。每个步骤都简单组合起来能干大事。链式调用是什么链式调用就是把多个处理步骤串成一条流水线。在LangChain里每一步是一个Runnable组件可以是提示词模板、模型、输出解析器、检索器甚至是你自己写的函数。组件之间用管道符连接数据从左往右流前一步的输出自动变成下一步的输入。如果你用过Unix管道cat file | grep keyword | sortLCEL的管道符思路一模一样。竖线左边的东西吐出来竖线右边的东西接住继续处理。每个组件只关心自己那一段逻辑不用知道上下游在干什么。这种解耦让你可以单独替换某个环节比如把GPT换成Claude或者把检索器换成另一个其余部分一行代码都不用改。LLMChain最基础的链LLMChain是LangChain最早期的链干的事情很简单把提示词模板、模型、输出解析器三件套绑在一起。老版本写法是直接实例化LLMChain类传一堆参数代码又长又难读。现在官方推荐用LCEL管道语法效果一样代码清爽很多。fromlangchain_openaiimportChatOpenAIfromlangchain_core.promptsimportChatPromptTemplatefromlangchain_core.output_parsersimportStrOutputParser modelChatOpenAI(modelgpt-4o-mini,temperature0)promptChatPromptTemplate.from_messages([(system,你是一个技术写作助手用简洁专业的风格写作),(human,请解释{concept}控制在200字以内),])chainprompt|model|StrOutputParser()resultchain.invoke({concept:向量检索})print(result)prompt、model、parser三个组件用竖线连起来。prompt把变量填进模板model生成回答StrOutputParser把AIMessage转成纯文本。这就是一条最基础的链你以后写的复杂链拆到底都是这个结构。顺序链把多步任务串起来很多任务一步搞不定。比如写文章先生成大纲再扩写正文这就需要顺序链。老版本LangChain有SequentialChain类专门干这个现在用LCEL写更直观。你把多条链用管道符接起来或者用RunnableLambda把自定义函数塞进流水线。fromlangchain_core.runnablesimportRunnableLambda# 第一步生成大纲outline_promptChatPromptTemplate.from_template(为题目{topic}生成一个三段式的文章大纲每段一句话)outline_chainoutline_prompt|model|StrOutputParser()# 第二步根据大纲扩写正文body_promptChatPromptTemplate.from_template(根据以下大纲写一篇完整文章大纲如下\n{outline})body_chainbody_prompt|model|StrOutputParser()# 串联大纲的输出喂给正文full_chain{outline:outline_chain}|body_chain resultfull_chain.invoke({topic:大模型Agent开发入门})print(result)这里有个细节容易卡住。第二步的输入参数名是outline第一步的输出默认是字符串直接喂进去body_chain不认。所以中间用字典包了一层把outline_chain的输出映射到outline这个key上。LCEL里这种用字典做字段重命名的写法很常见刚开始用容易懵多写两次就习惯了。路由链让流程分叉顺序链是一条道走到黑但有时候你得根据输入走不同路线。用户问编程问题走技术链问生活问题走闲聊链。这种分叉场景用路由链英文叫Router Chain。LCEL里可以用RunnableBranch实现路由也可以用RunnableLambda自己写分发逻辑。我更喜欢后者灵活一些调试也方便。# 三条不同风格的生成链tech_promptChatPromptTemplate.from_template(你是技术博主为{topic}写一段200字的技术分享)life_promptChatPromptTemplate.from_template(你是生活博主为{topic}写一段200字的生活随笔)general_promptChatPromptTemplate.from_template(为{topic}写一段200字的短文)tech_chaintech_prompt|model|StrOutputParser()life_chainlife_prompt|model|StrOutputParser()general_chaingeneral_prompt|model|StrOutputParser()defroute(info):topicinfo[topic]if代码intopicor编程intopicor程序intopic:returntech_chain.invoke({topic:topic})elif生活intopicor心情intopic:returnlife_chain.invoke({topic:topic})else:returngeneral_chain.invoke({topic:topic})routerRunnableLambda(route)print(router.invoke({topic:用Python写一个爬虫程序}))路由链还有一种更高级的玩法先跑一个分类链让模型判断类别再根据类别分发。这样灵活度更高但也更容易出幺蛾子。模型分错类后面整条链就跑歪了。建议在路由前加一层校验分类结果不在预期范围内就走默认链兜底。完整示例多步骤任务流水线把上面的东西拼起来做一个完整的内容生产流水线。输入一个主题先分类再路由到对应的生成链最后统一润色。fromlangchain_openaiimportChatOpenAIfromlangchain_core.promptsimportChatPromptTemplatefromlangchain_core.output_parsersimportStrOutputParserfromlangchain_core.runnablesimportRunnableLambda modelChatOpenAI(modelgpt-4o-mini,temperature0)# 分类链classify_promptChatPromptTemplate.from_template(判断话题{topic}的类别只输出技术或生活或通用)classify_chainclassify_prompt|model|StrOutputParser()# 三条生成链tech_promptChatPromptTemplate.from_template(你是技术博主为{topic}写一篇300字的技术分享)life_promptChatPromptTemplate.from_template(你是生活博主为{topic}写一篇300字的生活随笔)general_promptChatPromptTemplate.from_template(为{topic}写一篇300字的短文)tech_chaintech_prompt|model|StrOutputParser()life_chainlife_prompt|model|StrOutputParser()general_chaingeneral_prompt|model|StrOutputParser()defgenerate_by_category(info):category,topicinfo[category],info[topic]ifcategory技术:returntech_chain.invoke({topic:topic})elifcategory生活:returnlife_chain.invoke({topic:topic})returngeneral_chain.invoke({topic:topic})# 润色链polish_promptChatPromptTemplate.from_template(润色以下文章使其更通顺流畅保持原意\n{content})polish_chainpolish_prompt|model|StrOutputParser()# 组装完整流水线pipeline(RunnableLambda(lambdax:{topic:x[topic],category:classify_chain.invoke({topic:x[topic]}),})|RunnableLambda(generate_by_category)|RunnableLambda(lambdacontent:{content:content})|polish_chain)resultpipeline.invoke({topic:用Python写一个爬虫})print(result)跑起来你会看到先分类判定为技术再走技术生成链写出初稿最后润色链通顺文字。整条流水线一气呵成。调试这种多步链有个技巧。LCEL的每个组件都支持invoke单独调用你可以把链断开一步一步跑看每一步的输出对不对。哪一步出问题就修哪一步比一整条链跑完才发现错误好排查多了。一些踩坑心得写多步链最容易出的问题是中间数据格式不匹配。第一步输出字符串第二步期望字典直接接就报错。养成习惯每一步想清楚输入是什么、输出是什么必要时用RunnableLambda做格式转换。还有一个坑是token消耗。链越长调模型的次数越多token烧得越快。生产环境里要监控每一步的token用量发现某个环节特别费钱就优化提示词或者合并步骤。最后提一句错误处理。多步链中间任何一步崩了整条链就断了。LCEL组件抛异常会直接往上冒你可以在外层套try-except也可以用RunnableLambda在关键步骤做容错。比如分类链返回了乱码后续步骤拿到乱码只会越跑越离谱不如在这一步就拦住返回一个默认值让流程继续走。这篇用LCEL管道语法写了三种链基础的LLMChain、顺序执行的顺序链、分叉的路由链最后拼了一个完整的内容生产流水线。你会发现LCEL的管道写法贯穿始终竖线一接数据就流过去了。下一篇我们深入LCEL表达式语言本身看看管道符背后到底发生了什么Runnable接口还有哪些用法。
链式调用实战用LangChain构建多步骤复杂任务流水线上一篇把LangChain五个核心概念过了一遍最后提到要用LCEL写链式调用。这篇就来动手从最基础的链开始一步步搭到完整的多步骤任务流水线。先回答一个基本问题。为什么我们需要链。单次调用不够用你写第一个AI应用的时候大概率是这样干的。拼一段提示词调一次模型拿结果返回。一个来回搞定简单直接。但真实业务很少这么省心。用户让你写一篇技术博客你得先生成大纲大纲确认后再逐段扩写最后还要润色。用户问你一个专业问题你可能要先判断问题属于哪个领域再走对应的检索流程最后组织回答。这些场景一个模型调用根本搞不定。我自己踩过一个坑。做一个文档摘要功能一开始想偷懒一个提示词让模型又提炼要点又润色又排版。结果模型顾此失彼要点抓不全文字也不够通顺。拆成三步之后先提炼再润色最后排版每一步只干一件事质量立马好了一截。这就是链式调用的价值。把一个复杂任务拆成几个小步骤每个步骤用一个链处理前一步的输出喂给下一步。每个步骤都简单组合起来能干大事。链式调用是什么链式调用就是把多个处理步骤串成一条流水线。在LangChain里每一步是一个Runnable组件可以是提示词模板、模型、输出解析器、检索器甚至是你自己写的函数。组件之间用管道符连接数据从左往右流前一步的输出自动变成下一步的输入。如果你用过Unix管道cat file | grep keyword | sortLCEL的管道符思路一模一样。竖线左边的东西吐出来竖线右边的东西接住继续处理。每个组件只关心自己那一段逻辑不用知道上下游在干什么。这种解耦让你可以单独替换某个环节比如把GPT换成Claude或者把检索器换成另一个其余部分一行代码都不用改。LLMChain最基础的链LLMChain是LangChain最早期的链干的事情很简单把提示词模板、模型、输出解析器三件套绑在一起。老版本写法是直接实例化LLMChain类传一堆参数代码又长又难读。现在官方推荐用LCEL管道语法效果一样代码清爽很多。fromlangchain_openaiimportChatOpenAIfromlangchain_core.promptsimportChatPromptTemplatefromlangchain_core.output_parsersimportStrOutputParser modelChatOpenAI(modelgpt-4o-mini,temperature0)promptChatPromptTemplate.from_messages([(system,你是一个技术写作助手用简洁专业的风格写作),(human,请解释{concept}控制在200字以内),])chainprompt|model|StrOutputParser()resultchain.invoke({concept:向量检索})print(result)prompt、model、parser三个组件用竖线连起来。prompt把变量填进模板model生成回答StrOutputParser把AIMessage转成纯文本。这就是一条最基础的链你以后写的复杂链拆到底都是这个结构。顺序链把多步任务串起来很多任务一步搞不定。比如写文章先生成大纲再扩写正文这就需要顺序链。老版本LangChain有SequentialChain类专门干这个现在用LCEL写更直观。你把多条链用管道符接起来或者用RunnableLambda把自定义函数塞进流水线。fromlangchain_core.runnablesimportRunnableLambda# 第一步生成大纲outline_promptChatPromptTemplate.from_template(为题目{topic}生成一个三段式的文章大纲每段一句话)outline_chainoutline_prompt|model|StrOutputParser()# 第二步根据大纲扩写正文body_promptChatPromptTemplate.from_template(根据以下大纲写一篇完整文章大纲如下\n{outline})body_chainbody_prompt|model|StrOutputParser()# 串联大纲的输出喂给正文full_chain{outline:outline_chain}|body_chain resultfull_chain.invoke({topic:大模型Agent开发入门})print(result)这里有个细节容易卡住。第二步的输入参数名是outline第一步的输出默认是字符串直接喂进去body_chain不认。所以中间用字典包了一层把outline_chain的输出映射到outline这个key上。LCEL里这种用字典做字段重命名的写法很常见刚开始用容易懵多写两次就习惯了。路由链让流程分叉顺序链是一条道走到黑但有时候你得根据输入走不同路线。用户问编程问题走技术链问生活问题走闲聊链。这种分叉场景用路由链英文叫Router Chain。LCEL里可以用RunnableBranch实现路由也可以用RunnableLambda自己写分发逻辑。我更喜欢后者灵活一些调试也方便。# 三条不同风格的生成链tech_promptChatPromptTemplate.from_template(你是技术博主为{topic}写一段200字的技术分享)life_promptChatPromptTemplate.from_template(你是生活博主为{topic}写一段200字的生活随笔)general_promptChatPromptTemplate.from_template(为{topic}写一段200字的短文)tech_chaintech_prompt|model|StrOutputParser()life_chainlife_prompt|model|StrOutputParser()general_chaingeneral_prompt|model|StrOutputParser()defroute(info):topicinfo[topic]if代码intopicor编程intopicor程序intopic:returntech_chain.invoke({topic:topic})elif生活intopicor心情intopic:returnlife_chain.invoke({topic:topic})else:returngeneral_chain.invoke({topic:topic})routerRunnableLambda(route)print(router.invoke({topic:用Python写一个爬虫程序}))路由链还有一种更高级的玩法先跑一个分类链让模型判断类别再根据类别分发。这样灵活度更高但也更容易出幺蛾子。模型分错类后面整条链就跑歪了。建议在路由前加一层校验分类结果不在预期范围内就走默认链兜底。完整示例多步骤任务流水线把上面的东西拼起来做一个完整的内容生产流水线。输入一个主题先分类再路由到对应的生成链最后统一润色。fromlangchain_openaiimportChatOpenAIfromlangchain_core.promptsimportChatPromptTemplatefromlangchain_core.output_parsersimportStrOutputParserfromlangchain_core.runnablesimportRunnableLambda modelChatOpenAI(modelgpt-4o-mini,temperature0)# 分类链classify_promptChatPromptTemplate.from_template(判断话题{topic}的类别只输出技术或生活或通用)classify_chainclassify_prompt|model|StrOutputParser()# 三条生成链tech_promptChatPromptTemplate.from_template(你是技术博主为{topic}写一篇300字的技术分享)life_promptChatPromptTemplate.from_template(你是生活博主为{topic}写一篇300字的生活随笔)general_promptChatPromptTemplate.from_template(为{topic}写一篇300字的短文)tech_chaintech_prompt|model|StrOutputParser()life_chainlife_prompt|model|StrOutputParser()general_chaingeneral_prompt|model|StrOutputParser()defgenerate_by_category(info):category,topicinfo[category],info[topic]ifcategory技术:returntech_chain.invoke({topic:topic})elifcategory生活:returnlife_chain.invoke({topic:topic})returngeneral_chain.invoke({topic:topic})# 润色链polish_promptChatPromptTemplate.from_template(润色以下文章使其更通顺流畅保持原意\n{content})polish_chainpolish_prompt|model|StrOutputParser()# 组装完整流水线pipeline(RunnableLambda(lambdax:{topic:x[topic],category:classify_chain.invoke({topic:x[topic]}),})|RunnableLambda(generate_by_category)|RunnableLambda(lambdacontent:{content:content})|polish_chain)resultpipeline.invoke({topic:用Python写一个爬虫})print(result)跑起来你会看到先分类判定为技术再走技术生成链写出初稿最后润色链通顺文字。整条流水线一气呵成。调试这种多步链有个技巧。LCEL的每个组件都支持invoke单独调用你可以把链断开一步一步跑看每一步的输出对不对。哪一步出问题就修哪一步比一整条链跑完才发现错误好排查多了。一些踩坑心得写多步链最容易出的问题是中间数据格式不匹配。第一步输出字符串第二步期望字典直接接就报错。养成习惯每一步想清楚输入是什么、输出是什么必要时用RunnableLambda做格式转换。还有一个坑是token消耗。链越长调模型的次数越多token烧得越快。生产环境里要监控每一步的token用量发现某个环节特别费钱就优化提示词或者合并步骤。最后提一句错误处理。多步链中间任何一步崩了整条链就断了。LCEL组件抛异常会直接往上冒你可以在外层套try-except也可以用RunnableLambda在关键步骤做容错。比如分类链返回了乱码后续步骤拿到乱码只会越跑越离谱不如在这一步就拦住返回一个默认值让流程继续走。这篇用LCEL管道语法写了三种链基础的LLMChain、顺序执行的顺序链、分叉的路由链最后拼了一个完整的内容生产流水线。你会发现LCEL的管道写法贯穿始终竖线一接数据就流过去了。下一篇我们深入LCEL表达式语言本身看看管道符背后到底发生了什么Runnable接口还有哪些用法。
)
