★ FEATURED ARTICLE 刚开始用LangChain写应用的朋友多半都会遇到同一个问题模型单次返回的结果总是不尽如人意要么格式不对要么内容不完整要么就是一个简单的问答还好一旦涉及多步骤任务就手足无措。其实很多时候不是你提示词写得不行而是你缺少一个把模型调用组织起来的结构。这个结构在LangChain里就是Chains链。这篇指南就围绕链来展开聊清楚它解决了什么问题、由哪些核心组件构成、怎么从零搭一条可用的链以及实际项目中容易踩的坑适合刚接触LangChain的开发者参考也适合已经写过一些Demo但想理清设计思路的人。1. 链的核心设计思路为什么要用链1.1 单次调用模型的痛点先看一个很常见的业务需求根据用户输入的商品名生成一段带货文案。如果只调一次模型Prompt大概是这样的“请为商品XXX写一段带货文案”。模型确实会返回一段文案但你拿到的内容是随机的有时候是纯文本有时候带Markdown标题有时候还自己加了几句评价。你想要的是“标题卖点列表购买引导”这种稳定结构一次调用做不到。你可能会说提示词里写清楚格式不就行了确实可以但问题在于真实业务往往不止“写一段文案”这一个动作。你可能需要先生成商品卖点再基于卖点写文案再提炼一段SEO摘要再把摘要做成社交媒体的短文案。这种多阶段、有依赖关系的任务靠单次模型调用是没法完成的。链就是为了解决这类问题而存在的。它的本质很简单把多个模型调用和数据处理步骤按顺序、按条件组合成一个工作流。每个环节只做一件事上一环的输出作为下一环的输入。这种流水线式的设计让代码结构清晰、可调试、可复用。1.2 从Legacy Chain到LCELLangChain表达式语言如果你看LangChain的老教程会看到一堆类LLMChain、SimpleSequentialChain、RouterChain等等。这些被称为Legacy Chain传统链它们的写法很啰嗦而且不同链之间不能灵活组合。你现在打开LangChain的文档官方已经明确建议用LCELLangChain Expression Language来构建链。LCEL是一种声明式的组合语法核心就一个管道符|比如prompt | model | parser就构成了一条链。它的设计思路和Unix管道类似左边是数据源或处理单元右边是接收上一步输出的处理单元数据从左边流到右边。用LCEL的好处不只是写法简洁它还免费获得了一些高级特性流式返回模型边生成边输出、异步调用支持、并行执行、自动重试、以及每一步的可观测性可以轻松打印中间结果。这些特性在实际项目中非常重要传统链需要额外代码才能实现。所以尽管你的项目标题是“链完全指南”但我建议你把重心放在LCEL上而不是去背那些即将被淘汰的类。理解链的核心概念再用LCEL去表达这才是2024年以后写LangChain应用的正确姿势。1.3 链的设计原则组件化思维链的设计遵循一个很朴素的原则把任务拆成可独立替换的组件然后用编排逻辑把它们串起来。打个比方做一道菜你不需要把一个“做菜模型”塞进一个黑盒里而是“切菜”、“炒制”、“调味”分别处理每一步都可以调优和替换。在LangChain里一个最基础的链由三个组件构成提示词模板PromptTemplate负责把输入变量格式化成模型能理解的指令模型LLM/ChatModel负责生成内容输出解析器OutputParser把模型返回的文本转换成结构化数据。三者的职责完全分开。这意味着你可以单独改提示词而不用动代码逻辑可以换不同的模型服务商而不用改业务代码也可以替换解析策略来适应用户需求的变化。这种组件化思维是链设计中最核心的思想后面所有复杂的链都建立在这三个基础组件之上。2. 链的核心组件拆解与参数理解2.1 提示词模板PromptTemplate链的输入入口在LCEL表达式中PromptTemplate通常就是链条的第一环。它做的事很简单把你的输入变量填充到一段预定义的模板里。LangChain支持f-string风格的标准模板也支持Jinja2风格。from langchain_core.prompts import PromptTemplate # f-string风格最常用 template PromptTemplate.from_template( 你是资深带货文案专家。请给商品「{product}」写一段250字以内的种草文案 包含一个吸引人的开头、3个核心卖点、以及结尾购买引导。 ) # 也可以显式声明输入变量更严谨 template_with_vars PromptTemplate( input_variables[product, audience], template为面向「{audience}」的消费者撰写「{product}」的营销文案\n{format_instructions}, )同时你还可以用partial_variables部分变量来预置一些不会频繁变化的变量比如当前日期、固定的品牌名称或格式指令这样运行时就不必每次传一遍。在实际使用中模板的输入变量名与后续链中用到的变量名必须严格一致。很多人刚开始写链一个低级错误就是模板里写{product}但传入链的字典键是product_name导致运行时报“Missing variables”错误。建议从一开始就统一变量命名习惯维护起来会轻松很多。2.2 模型接口与关键参数温度、最大长度等的取舍LangChain的核心抽象之一是把不同提供商的模型统一为同一个接口。在LCEL链中你可以用ChatOpenAI、ChatAnthropic或者本地模型封装。它们的共同点是输入是消息列表输出是AIMessage对象。from langchain_openai import ChatOpenAI # 注意生产环境不要硬编码密钥通过环境变量读取 llm ChatOpenAI( modelgpt-4o-mini, temperature0.7, max_tokens2048, )从实际调用效果来说temperature对创作类任务影响很大做头脑风暴、写营销文案可以把temperature调高到0.7~0.9让输出更有随机性抽取结构化信息、做数据分类、写代码则建议调到0.1甚至0让输出更稳定。max_tokens要根据输出长度预估链式调用中尤其要预留足够的token余量因为中间步骤的输出会成为下一步输入的一部分token消耗会叠加。值得提醒的是同一个模型参数在不同提供商上名称有差异比如有些平台叫max_output_tokens有些叫max_tokens你换模型服务商时容易踩坑。LangChain试图帮你屏蔽差异但底层参数映射并不总是完美遇到“参数不识别”之类报错时直接去查对应封装类的源码或文档比自己瞎猜更快。2.3 输出解析器OutputParser与结构化输出如果说模型是大脑输出解析器就是那双手把你需要的结果从模型返回的长篇文本里“取”出来。最常见的是StrOutputParser它只把AIMessage的文本内容取出来转成字符串如果模型返回的是JSON就用JsonOutputParser如果希望模型按Pydantic模型返回可以用PydanticOutputParser。from langchain_core.output_parsers import StrOutputParser from langchain_core.output_parsers import JsonOutputParser str_parser StrOutputParser() json_parser JsonOutputParser()对于结构化输出LangChain在最新版本里推荐直接用with_structured_output()方法配合模型本身的Function Calling能力比单纯靠提示词约束JSON格式要可靠得多。from pydantic import BaseModel, Field class ProductCopy(BaseModel): title: str Field(description文案标题) selling_points: list[str] Field(description核心卖点列表至少3条) call_to_action: str Field(description结尾购买引导语) structured_llm llm.with_structured_output(ProductCopy) result structured_llm.invoke(写一个降噪耳机的带货文案) print(result.title) # 直接访问属性 print(result.selling_points) # 列表类型不需要再处理字符串有了这个能力输出解析器的容错率大幅提升。如果模型返回不了合法JSON传统解析器会直接抛错导致链终止而with_structured_output依赖模型的原生工具调用能力绝大多数情况下能返回干净的结构化数据。不过要注意这个能力需要模型供应商提供原生工具接口如果你用的是本地开源模型且没启用工具调用它可能自动退化为提示词约束模式这时候JSON解析失败的概率又会回升。2.4 记忆Memory和检索器Retriever作为链的“外挂”基础的提示词模板、模型、解析器三元组只适合无状态的一次性转换。但如果你的链是多轮对话比如客服机器人就需要记忆组件来保存历史会话如果你的任务是知识库问答就需要检索器从外部文档里找上下文。LCEL中接入记忆常用的方式是RunnableWithMessageHistory。它会根据会话ID从存储里加载历史消息自动拼接到本轮输入之前然后传给模型。这个封装用起来很方便但有一个要留意的点输入的提示词模板需要预留{history}变量位置否则历史消息不知道插在哪里。检索器接入链的方式也很直白——检索器本身是一个Runnable可以作为链的一环先运行拿到相关文档后在下一步拼接进提示词。from langchain_core.runnables import RunnablePassthrough # retriever是已经构建好的向量检索器 retrieval_chain ( {context: retriever, question: RunnablePassthrough()} | prompt | llm | str_parser )这里RunnablePassthrough()是一个透传组件它会把用户的原始输入原封不动传给下一步而检索器负责把查询转为文档列表。这样链的输入变量就有两个来源检索结果和用户原文串联逻辑一目了然。3. 实操过程从简单链到复杂链的完整实现3.1 第一类链单步转换链LCEL入门五连拍我们先从一条最简单的链开始用户输入商品名称模型返回卖点列表。这虽然简单但麻雀虽小五脏俱全你看到的就是LCEL最标准的范式。from langchain_core.prompts import PromptTemplate from langchain_openai import ChatOpenAI from langchain_core.output_parsers import JsonOutputParser prompt PromptTemplate.from_template( 你是电商产品专家。请分析商品「{product}」的核心卖点 以JSON数组形式返回每个元素包含卖点标题和说明。 ) llm ChatOpenAI(modelgpt-4o-mini, temperature0.3) chain prompt | llm | JsonOutputParser() result chain.invoke({product: 无线降噪耳机})你可以一行一行地读这段链prompt把product变量格式化成一个完整的指令llm接收这个指令并生成回复JsonOutputParser把回复解析成Python列表。最后你得到一个可以直接用于下游业务的list。链的可组合性在这里已经展现每一步都只依赖前一步的输出结构和链条其他部分完全解耦。如果你想知道每一步的中间输出LCEL提供了简便的调试手段from langchain_core.runnables import RunnableLambda def debug_step(step_name): return RunnableLambda(lambda x: print(f--- {step_name} ---) or x) chain prompt | debug_step(prompt输出) | llm | debug_step(llm输出) | JsonOutputParser()用RunnableLambda包一个打印函数进去就能在指定环节打印中间结果。这种排查方式比官方提供的verbose选项更灵活想加在哪一步都行而且不需要改主逻辑。3.2 第二类链顺序链SequentialChain让输出层层递进单步转换适用于简单场景但真实业务中的任务往往有依赖关系。比如做一个内容自动生成器先根据主题生成大纲再基于大纲写正文最后提炼SEO摘要。这就是典型的SequentialChain场景。from langchain_core.runnables import RunnablePassthrough outline_prompt PromptTemplate.from_template( 为文章主题「{topic}」生成一个3层结构的文章大纲每层用编号列表表示。 ) article_prompt PromptTemplate.from_template( 请根据以下大纲扩写一篇完整的干货型文章。\n大纲\n{outline}\n\n 要求逻辑清晰每节不少于200字使用实践者口吻。 ) seo_prompt PromptTemplate.from_template( 请为下面这篇文章提炼一段SEO摘要包含关键词建议和搜索意图分析。\n文章\n{article} ) llm ChatOpenAI(modelgpt-4o, temperature0.5) chain ( RunnablePassthrough.assign(outlineoutline_prompt | llm | StrOutputParser()) .assign(articlearticle_prompt | llm | StrOutputParser()) .assign(seoseo_prompt | llm | StrOutputParser()) ) result chain.invoke({topic: 链式调用的设计模式})这里使用了RunnablePassthrough.assign()方法它的作用是每执行一步把新变量塞进结果字典同时保留之前的变量。这保证了后一步可以访问前面所有步骤的结果比如article_prompt能同时看到原始topic和生成的outline。在较老版本的LangChain中SequentialChain需要额外管理input_variables和output_variables很容易因为某个变量没声明而报错。LCEL的.assign()链式调用让变量传递变得异常直观这也是我强烈建议新项目直接用LCEL的原因。3.3 第三类链路由链RouterChain让请求按条件分发不是所有输入都适合走同一条处理流程。比如用户想查天气时直接查天气API想闲聊时走闲聊模型想分析图片时走多模态模型。如果全部塞进一个提示词里效果会很差。路由链RouterChain就是解决这个分发问题的。在LCEL中路由通常由RunnableBranch实现。它会根据一个路由函数的结果选择不同的下游链。from langchain_core.runnables import RunnableBranch, RunnableLambda def route_by_keyword(input_dict): text input_dict[text] if 天气 in text: return weather elif 写 in text or 文案 in text: return copywriting else: return chat weather_chain prompt_weather | llm | StrOutputParser() copywriting_chain prompt_copywriting | llm | StrOutputParser() chat_chain prompt_chat | llm | StrOutputParser() branch RunnableBranch( (lambda x: route_by_keyword(x) weather, weather_chain), (lambda x: route_by_keyword(x) copywriting, copywriting_chain), chat_chain # 默认分支兜底 ) router_chain {text: RunnablePassthrough()} | branch这样一次链的调用就同时完成了意图识别和任务执行。RunnableBranch按顺序从上到下匹配每个条件第一个返回True的分支会被执行如果所有条件都不满足就走最后一个默认分支。这里的兜底分支很重要否则不在规则内的输入会直接抛错。从维护角度说路由逻辑独立为函数更适合测试你可以单独写几个单元测试来验证关键词判断是否正确而不是每次都调用整个路由链。这样设计的好处是路由条件的变更不需要改业务链只需改路由函数。3.4 第四类链自定义链与特定场景拼装LCEL的管道运算符组合了内置组件但在某些场景下你需要自定义逻辑。比如中间要调用一个标准化Python函数、调一下数据库或者把输出转成特定格式。这时候用RunnableLambda包一个函数即可。from langchain_core.runnables import RunnableLambda def normalize_copy(text: str) - str: # 去除多余空格、限制长度 clean .join(text.split()) return clean[:500] normalization_step RunnableLambda(normalize_copy) chain prompt | llm | StrOutputParser() | normalization_step如果你需要更复杂的自定义行为比如包含状态下转换的链可以继承Runnable类并实现invoke、stream等方法。但绝大多数需求用RunnableLambda就够了没必要为了形式上的“自定义”去写一整个类除非你需要复用很多次或者逻辑非常复杂。在LLM应用开发中很多时候链的一个环节需要调用外部API来获取实时数据比如比价、查库存、查询订单物流。你可以在RunnableLambda函数里调用requests库把返回值拼装到提示词里让模型基于实时数据来回答。这种“工具增强型链”在实际业务中非常常见也是从“会跑Demo”到“能上生产”的重要一步。3.5 实操记录一次完整的链开发过程示例我以前做过一个内部内容助手其中一条链的开发过程比较有代表性分享出来供你参考。接到需求时业务方说“用户输入一个主题我们要生成一份完整的内容日历包含七个内容方向、每个方向的文案示例和发布建议。”第一版实现很粗暴把所有要求塞进一个超长Prompt调一次模型。结果是输出结构经常乱有时七个方向只有四个有时方向之间的格式不统一用起来非常痛苦。第二版改成链式调用环节1把主题拆成“用户人群分析”输出结构化的人群特征列表。环节2基于人群特征生成7个内容方向标题。环节3为每个方向生成文案示例这里用了RunnableParallel并行处理把7个方向分别交给模型并行生成耗时没有线性增长。环节4把所有结果拼接成一个完整的内容日历并输出JSON格式。代码结构大致是from langchain_core.runnables import RunnableParallel generate_directions prompt_directions | llm | JsonOutputParser() generate_copy_for_direction ( prompt_one_copy | llm | StrOutputParser() ) parallel_copy_gen RunnableParallel( **{fdirection_{i}: generate_copy_for_direction for i in range(7)} ) calendar_chain ( RunnablePassthrough.assign(directionsgenerate_directions) .assign(copiesparallel_copy_gen) | prompt_calendar | llm | JsonOutputParser() )如果用传统写法第三步的7个方向只能用循环串行调用总耗时等于7倍的模型响应时间。而RunnableParallel可以让这7个生成任务并行执行只要API支持并发请求总时间就近似等于单次最慢的那个任务。这种优化对生产体验影响非常明显。当然并行调用要留意API的速率限制。如果你用的是某模型平台的免费额度并发请求很容易触发限流。个人经验是先小规模测试平台的并发上限再决定并行度别一上来就拉满。4. 常见问题与排查技巧实录4.1 模型输出结构化数据失败解析器频繁抛错这是链式开发中最容易遇到的状况。你让模型返回JSON它给你包了一段Markdown代码块你让模型列三个卖点它给你写了五句话加一段感慨。解决思路有优先级首选使用with_structured_output()配合Pydantic模型让模型原生按工具格式返回基本能规避大部分解析问题。次选在提示词里加入{format_instructions}这是PydanticOutputParser自带的格式指令它会把字段说明和JSON示例一并注入提示词。兜底编写重试逻辑捕获解析异常后自动让模型重新生成一次。from langchain_core.output_parsers import PydanticOutputParser from langchain_core.pydantic_v1 import BaseModel, Field class CopyResult(BaseModel): title: str Field(description文案标题) points: list[str] Field(description卖点列表) parser PydanticOutputParser(pydantic_objectCopyResult) prompt PromptTemplate( template完成任务并严格按格式输出。\n{format_instructions}\n任务{task}, input_variables[task], partial_variables{format_instructions: parser.get_format_instructions()}, )即便做了这些你仍要接受一个现实模型不是数据库偶尔出错是常态。生产级别的链必须在解析层之外再包一层校验逻辑如果校验失败就触发重试或者人工兜底而不是直接抛异常给用户。4.2 链式调用的上下文丢失中间输出被下一环节忽略顺序链中常见的第二个坑是“上下文丢失”。比如你先让模型总结文章再把总结喂给下一个模型让它扩写结果扩写的内容完全没有参考原文细节看起来像是从零开始写的。原因通常是你的中间输出太短或者被截断了。模型在下一步看到的只是上一步的摘要而摘要本身可能丢失了原文中的关键事实。解决办法是把原始文本和中间摘要一起传给下一步让模型同时看到原始材料和处理结果自己判断怎么使用。pass_through_original RunnablePassthrough.assign( original_textlambda x: x[article_text] )这种“永远保留原始输入”的习惯在写多级链时尤其重要。一旦开始做两级以上的转换尽量别丢弃最初的数据因为你不知道后面的环节需不需要。4.3 调试与可观测性用trace_log快速定位断点链步骤多了以后靠“打印在终端里看不出来是谁输错”的方式会效率很低。LangChain提供了两种主要调试手段第一种在LCEL链中插入RunnableLambda打印节点在关键步骤前后打印输入输出。适合即时调试修改起来也方便。第二种使用LangSmith或Langfuse这类可观测性平台把每次链调用的完整trace记录下来每一步的输入输出、耗时、token消耗和成本都可视化。适合线上排查问题。个人经验是开发阶段用打印节点部署后靠平台追踪。两者并不冲突。线上如果出现某个环节的模型返回频繁异常你能在trace里看到具体是哪个环节、哪次调用出了问题而不需要靠猜测去复现。4.4 性能与成本链越长token消耗越大链的代价是token消耗每一步模型的输入输出都是token后一步的输入还包含了前一步的输出所以链条越长实际消耗增长越明显。一个四步的链调用的总token可能是单步调用的4到8倍成本控制不好就会被账单吓一跳。控制成本的方法有几类能并行的一步并行缩短时间能缓存的重复杂用缓存langchain_core.caches不需要模型处理的部分用规则代码处理省掉模型调用。最核心的思路是模型只做必须靠语义理解才能完成的事字符串拼接、格式整理、条件判断这些尽量交给代码。4.5 一个完整的问题排查速查表症状可能原因排查与修复提示“Missing some input variables”模板变量名与传入键不一致检查PromptTemplate的input_variables与实际字典键输出解析器抛JSONDecodeError模型返回了Markdown或多余文本改用with_structured_output在提示词强调“只输出JSON”加重试逻辑链的输出内容偏离主题中间步骤丢失了关键上下文保留原始输入同步传给后环节精简中间摘要调用报速率限制并行度过高或配额不足降低并发数增加重试和退避机制模型生成停不下来max_tokens设置过大且模型自由发挥控制max_tokens强化输出约束结果时而好用时而崩temperature过高结构化处理降低temperature甚至设为0流式输出失效链中存在非流式步骤避免在链中间用阻塞式RunnableLambda检查模型是否支持流式5. 链的工程化实践从Demo到生产环境5.1 什么时候该用链什么时候别用链链不是银弹。如果你的任务只需要一次简单问答那就没必要套上链的概念徒增复杂度。链真正发挥价值的场景是任务包含多个步骤、步骤之间有明确的数据依赖、每步的输入输出可以被结构化描述。举个例子一个“周报生成器”就很适合用链先收集工作事项再按项目分类再生成周报正文最后提炼本周总结。每个步骤输入输出清晰谁先谁后明确这就是链的舒适区。反之如果你的任务没有固定顺序需要模型自己决定下一步干什么比如“帮我想一个方案并不断自我评估修正”那么智能体Agent比链更合适。链是确定性的流水线智能体则更接近自主决策。很多人一上来就用Agent结果发现不可控其实他们的问题用一条链就解决了。5.2 链的粒度设计太粗难调太细难维护链的粒度是工程上最容易纠结的问题。步骤拆得太粗一个步骤承担了过多职责出问题不好定位拆得太细环节很多、变量名满天飞维护起来也很累。我自己的实践标准是以“一个环节是否可以被独立测试和复用”为准绳。如果一个环节能独立回答“给什么输入、产出什么输出、失败时如何降级”那它就适合作为一个链节点。如果一个环节拆开后无法单独使用必须依赖别的环节才能有意义说明拆得过细了。举个例子内容生成器中“生成大纲”和“生成正文”适合拆开因为大纲本身可以用来做目录预览。但“给正文添加标题”和“给正文统一格式”就应该合并因为单独看没什么意义。5.3 测试链的几种实用方法链的测试分四个层次组件级测试单独测PromptTemplate渲染结果、单独测输出解析器能否解析预期格式。单步测试固定输入验证链中某个环节的输出是否达标。全链测试用真实业务输入跑完整调用核对最终输出结构和内容质量。回归测试把历史典型输入保存成测试集每次改动链后批量跑一遍看有没有退化。# 简单的链测试示例用pytest风格组织 def test_copywriting_chain(): chain build_copywriting_chain() result chain.invoke({product: 无线鼠标}) assert title in result assert isinstance(result[selling_points], list) assert len(result[selling_points]) 3不要小看回归测试链这种东西改一个提示词可能全链路行为都变。我吃过几次亏改了某个环节的提示词格式化指令结果下游的解析器全挂了。从那以后我养成了跑回归测试的习惯成本不高但救命。5.4 版本管理与提示词迭代提示词本质上就是代码的一部分应该纳入版本管理。实际项目里可以用每个提示词单独建一个模块文件集中管理甚至可以把提示词写进配置文件方便业务人员参与调整。一个我自己很推荐的做法给提示词模板加版本号和变更记录说明比如v3_short_tone代表“第三版短句风格”。这样当同事反馈“效果比上周差”时能快速定位是哪一版提示词导致的而不需要翻聊天记录。5.5 接入生产前必须考虑的三件事第一超时与重试。模型API网络不稳定是常态链的每步都可能超时。生产级链必须为每一步配置超时时间和重试次数并且考虑重试后输出一致性。第二成本控制。把每次链调用的token消耗记录到日志设置异常报警比如单次调用超过多少token就告警。否则业务量上来之后你根本不知道钱花在哪里。第三安全与内容过滤。链的输入输出最好经过一层安全过滤避免用户输入恶意内容或模型输出不合规内容。这层过滤用规则或一个小模型做前置/后置校验虽然简单但在生产环境中很必要。6. 我的一些实际体会我用了很长一段时间链和Agent最深的感触是链式的确定性比智能体的自由发挥更适合生产任务。在你对问题域有足够理解后优先把任务流程显式地画出来再考虑用链去实现而不是把所有逻辑都抛给模型自己去编。链的每一条管道都是你对业务的一次思考抽象这个过程本身就在帮你理清产品需求。另外LangChain的版本迭代速度很快文档也不断变化。写这条链的代码在我这边运行正常但当你看到这篇文章时API可能已经发生了小范围变动。学习方法上我建议以理解核心抽象Prompts、Models、OutputParsers、Runnables、LCEL为主这些底层概念是稳定的具体类名和位置则随框架演进保持更新即可。最后分享一个小技巧当你开始觉得一条链变得臃肿、变量相互纠缠时别急着继续加环节先把链画成一张流程草图纸笔就行。把每个环节的输入、输出、依赖关系标注出来大多数情况下你会发现某些环节可以合并某些环节可以并行某些环节其实根本不需要模型参与。这一步花不了十分钟但能让你的链清晰不少也避免后面维护时一边翻代码一边叹气。
阅读完成 · 觉得有帮助？
