★ FEATURED ARTICLE 看到LangChain 进阶这种标题我估计点进来的朋友多半已经跑通过基础 Demo知道怎么调大模型接口、会写 PromptTemplate、也拼过最简单的链。但这个阶段的痛点往往不是不会用而是用得不明白——为什么链要这样串输出解析器到底是干嘛的为什么网上教程里那一堆括号和尖括号看着头晕这篇文章我就把这些事情掰开讲清楚重点放在 Chain 链的组织方式和输出解析器的工作原理上配合可以直接抄走的代码和踩坑记录。先说清楚这篇文章适合谁已经跑通过 LangChain 基础流程、想把手上的脚本改造成更工程化结构的开发者或者被链式调用结构化输出这些概念卡住的初学者。核心内容围绕三段展开Chain 链的设计逻辑与典型形态、输出解析器的原理与选型、以及两者配合时的稳定性问题。至于什么是 LangChain、怎么装环境这类基础内容默认你已经会了不会浪费篇幅。1. Chain 链真正要解决的是什么问题很多教程把 Chain 讲得很玄乎好像它是某种高深莫测的人工智能架构。拆穿了看Chain 就是一个有向的、按顺序执行的任务流水线——前一步的输出变成后一步的输入仅此而已。你用 Python 手写一个函数调用b(a(x))本质上也是在搭一条链条只不过 LangChain 把这种组合方式规范成了框架层面的协议并且顺手解决了重试、流式输出、中间结果可视化这些工程问题。1.1 从函数嵌套到标准化链路的演进逻辑最早大家写大模型应用代码大概是这样的result_a llm.invoke(prompt_a.format(user_inputuser_input)) result_b llm.invoke(prompt_b.format(contextresult_a)) result_c llm.invoke(prompt_c.format(dataresult_b))这种写法的问题不明显直到你需要把第 2 步的结果同时喂给第 3 步和第 4 步、在任意一步失败时重试、打印每一步的中间结果做调试、或者把整条链暴露成统一的接口。一旦需求变复杂手工维护的状态变量和异常处理就会失控。我见过一个项目15 个子任务串在一起中间结果到处传最后改参数的时候牵一发动全身三天两头出 bug。LangChain 的做法是定义一个统一的 Runnable 协议任何对象只要实现了invoke/batch/stream这几个方法就可以参与组链。提示词模板、模型封装、输出解析器、普通 Python 函数统统可以视为 Runnable然后用管道符|像拼乐高一样拼起来。1.2 LCEL 的管道语义一个看得见的流水线你大概已经见过了这种写法chain prompt | llm | parser拆开看每一步发生了什么prompt接收原始输入格式化成带变量的消息列表llm收下消息返回模型输出parser拿到模型输出解析成结构化对象。数据从左边流到右边每一步只对上一步的结果负责。这个设计最舒服的地方不是少写几行代码,而是链的每一环都可以被单独替换和测试。我实际开发中经常这么干先把prompt | llm跑通看原始输出对不对再单独测试parser手动喂几段奇形怪状的文本进去看解析器扛不扛得住最后再拼成完整链条。按环节拆着调比整条链一起跑 debug 高效得多。这里要提醒一句看到链管道的时候别被链这个名字带偏。它不是把多个大模型调用强绑在一起而是把大模型调用和其他类型的数据处理步骤统一进了一个可组合的框架。你完全可以把链条里的一环换成普通 Python 函数——比如数据清洗、格式转换、条件分支——这才是 LCEL 真正的价值所在。LangChain 官方在演进中不断强调 LCEL 取代旧版LLMChain类不是炫技而是因为这种一切皆 Runnable的统一抽象确实更符合工程需求。旧版链类存在太多私有的内部状态管理逻辑一旦要混合自定义代码就特别别扭。2. 顺序链与路由链两种最常见的链型实战拆解如果你的应用只有一个大模型调用老实说根本不需要 LangChain。真正让它发挥作用的场景是你有多个步骤、多条路径或者需要把非模型的逻辑也卷进来。我平时用得最多的链型有两种顺序执行链和条件路由链。2.1 顺序链串联多步任务的标准姿势顺序链适合输入要经过多轮加工每轮都调用大模型或函数的场景。典型例子先用模型判断一段客服工单的紧急程度再生成处理建议最后整理成结构化报表。先看用最新推荐写法LCEL 方式实现的顺序链这里我拿一个SQL 问题诊断的例子来说from langchain.prompts import PromptTemplate from langchain_core.runnables import RunnableLambda from langchain_core.output_parsers import StrOutputParser # 第一步识别问题类型 identify_prompt PromptTemplate.from_template( 你是一名数据库工程师。判断下面这个问题属于语法错误、性能优化、权限问题、还是其他。只回答类别名称。\n问题{query} ) # 第二步根据类型给出诊断建议 diagnose_prompt PromptTemplate.from_template( 问题类型为{category}。请针对原问题给出诊断建议要求分步骤说明。\n原问题{query} ) def extract_category(result: str) - dict: 把第一步的结果整理成第二步需要的输入字典。关键点LCEL 链中每个节点要么返回字符串要么返回字典后续节点会按照模板变量名自动匹配。 return {category: result.strip(), query: result_holder.get(query, )}等一下上面这段写岔了实际工程里你没法在函数里这样取result_holder。正确做法是让第一步的输出包含完整信息或者利用 LCEL 的itemgetter将多个输入路由到不同节点。建议这样写from operator import itemgetter chain ( {category: identify_prompt | llm | StrOutputParser(), query: itemgetter(query)} | diagnose_prompt | llm | StrOutputParser() )这个写法的核心是利用字典构造器把原始输入query原样传给后续节点同时把第一步的识别结果作为category传入。跑起来后链路逻辑非常清晰左边两个分支并行起步汇合后进入第二步提示词。很多新手在这里卡住是想不通怎么让第二步同时拿到原始输入和上一步输出。答案就是上面这招——用itemgetter和字典组合器显式声明每条数据的流向。如果你的项目还在维护旧代码可能会碰到SequentialChain这个类。我的建议是新代码别再用它了LCEL 的.pipe()或字典组合器功能完全覆盖旧类的能力而且中间状态管理更透明。旧类必须显式声明input_variables和output_variables改一个变量名要同步改好几处还要担心输出变量覆盖问题我在一个老项目里就是因为intermediate_steps字段名冲突排查了整整一个下午。2.2 路由链让大模型当一次分发器路由链RouterChain解决的问题是用户的请求可能有多种类型每种类型需要走完全不同的处理流程。比如一个技术助手接到的消息可能是 SQL 问题、代码调试、闲聊寒暄对应的处理链完全不同。我惯用的方案是两步走先用一个轻量模型做意图分类然后根据分类结果路由到具体链。代码骨架router_prompt PromptTemplate.from_template( 请把用户的请求分类到以下类别之一sql、python、chat。只输出类别名称。\n请求{input} ) def route(info: dict): category llm.invoke(router_prompt.format(inputinfo[input])).content.strip() if category sql: return sql_chain elif category python: return python_chain else: return chat_chain full_chain RunnableLambda(route)路由链的价值是在不增加用户感知延迟的前提下把不同质量要求的任务分给不同模型。比如 SQL 类请求对准确性要求高可以走能力更强的模型闲聊类请求追求响应速度可以走更轻量的模型。模型路由和任务路由结合起来成本能降下来不少——我手上有个内部工具这样分流后整体 API 费用少了大约三成因为大部分流量都是闲聊都被送去便宜的小模型了。这里有个容易翻车的细节分类提示词里给的类别不要太多不要超过五个。类别一多小模型的分类准确率断崖式下降。我测试过五个类别以上的路由准确率大概只有七成多属于不可用状态。这时候要么换能力更强的模型做分类要么改成用嵌入向量相似度匹配来分类后者在很多场景下反而更稳定。2.3 自定义链的边界何时值得自己动手写 Runnable框架自带的链类覆盖了 80% 的场景但总有例外你要做复杂的条件判断、需要访问外部数据库、或者要在链的中间插入文件 I/O 操作。这时候实现一个自定义 Runnable 是最干净的做法from langchain_core.runnables import Runnable, RunnableConfig from typing import Any, Iterator class ConfigCheckRunnable(Runnable): def invoke(self, input: Any, config: RunnableConfig | None None) - Any: # 这里可以写任意逻辑比如从配置中心拉参数 if env not in input: raise ValueError(缺少 env 参数) return {checked_config: input[env], original: input}写自定义 Runnable 最容易犯的错误是忘记返回结构统一的字典。既然下游节点按变量名取数据你返回的数据就必须包含下游模板需要的所有变量。我在 review 同事代码时见过很多次上游返回的 key 叫solution下游模板用的变量名是answer运行时报 KeyError 后他怎么都想不通哪里连错了。解决这个问题的笨办法是在自定义 Runnable 里最后打一行日志把返回的字典 key 列表打出来排查效率会高很多。3. 输出解析器让大模型从说人话到交数据如果你只需要把大模型的输出直接呈现给用户用StrOutputParser就够了。可一旦用户内容是给下游程序用的你需要的是稳定的数据结构而不是一段漂亮的文字。大模型的天然输出是自然语言里面经常带解释性前缀、markdown 标记、多余的空行——比如你让它输出 JSON它偏偏回你好的这是您需要的 JSON\njson\n{...}\n。这时候输出解析器就派上用场了。3.1 解析器的双段机制格式指令与结果解析LangChain 的输出解析器OutputParser底层机制是两段式配合get_format_instructions()生成一段格式说明文本被注入到 Prompt 里指导模型按指定结构输出。parse()接收模型原始输出按照既定规则解析成目标结构。大多数解析器还支持parse_with_prompt()此方法会在解析失败时把原始 Prompt 也传给解析器部分解析器可以借助用户原本的需求 当前回答自动修正小错误。比如模型忘了把某个枚举值大写解析器可以结合原始指令猜出正确写法。以 Pydantic 解析器为例它的get_format_instructions()会输出一段包含 JSON 字段说明、类型要求和示例的文本比如result: string这种。这段文本被塞进提示词尾部后模型按图索骥、输出对应结构的概率就大大提升了。我直接说结论你无法依靠请输出 JSON这句自然语言保证模型每次都乖乖交规整 JSON。原因是大模型的指令遵循存在随机性同一个提示词这次管用、下次就给你夹带私货。而解析器干的事情不是提高概率,而是把格式要求显式转化为模型更容易遵循的规范指令序列并在输出不合格时明确报错——让你在开发期就发现问题而不是等程序跑到生产环境里突然打出一堆 undefined。3.2 PydanticOutputParser最常用的结构化解析方案直接看完整可跑的代码。假设我要让模型抽取一段技术问题描述里的关键组件from langchain.output_parsers import PydanticOutputParser from langchain.prompts import PromptTemplate from langchain_openai import ChatOpenAI from pydantic import BaseModel, Field class IssueReport(BaseModel): issue_type: str Field(description问题类型bug或feature) component: str Field(description涉及的功能模块名称) description: str Field(description问题详细描述) parser PydanticOutputParser(pydantic_objectIssueReport) prompt PromptTemplate( template请分析下面的技术工单输出结构化信息。\n工单内容{input}\n{format_instructions}, input_variables[input], partial_variables{format_instructions: parser.get_format_instructions()}, ) llm ChatOpenAI(modelgpt-4o-mini, temperature0) chain prompt | llm | parser report chain.invoke({input: 用户在登录页点击验证码后页面崩溃报错提示 XX 模块为空}) print(report.issue_type) # bug print(report.component) # 登录页验证码组件这套方案的好处是模型输出先被解析成 Pydantic 对象字段类型、必填约束在解析阶段就强制执行。如果模型漏了某个字段解析器会抛OutputParserException提示缺了什么你可以在异常处理里写重试逻辑或者告警。温度参数要特别注意结构化输出场景建议把temperature调到 0 或接近 0。原因很简单温度参数控制采样随机性温度越高、模型越容易在格式和内容之间发挥创意——你以为它在努力给出正确答案其实它在努力把内容写得更像人话结果就是 JSON 格式崩坏。网上很多解析失败的案例排查到最后都是温度参数没调低。3.3 从 Pydantic 解析器到函数调用两条路线的正确姿势LangChain 生态里还有一条路线很多人没注意到就是通过Instructor这类工具直接基于函数调用function calling完成结构化输出。它的原理不是提示模型按格式输出而是把 Pydantic 模型定义转换成函数入参 Schema由模型内部原生的工具调用机制来决定如何填充结构化参数。这里我展开讲讲两条路线的核心区别因为太多人混着用最后不知道问题出在哪个环节。提示词约束路线的本质是求模型配合。模型的注意力机制要同时处理理解用户问题和遵循输出格式两件事在字段嵌套深、枚举值多的时候模型极容易为了内容准确性而牺牲格式准确性——它可能所有字段都对但 JSON 里多了个引号、少了个逗号、或者在 JSON 之前多写了好的没问题。PydanticOutputParser 里内置的自动修正逻辑结合原始 prompt 重新推断能救一部分但救不了所有情况。函数调用路线的本质是走专用通道。模型在预训练和微调阶段就被教会了如何调用工具输出 Tool Call 的时候走的是一条不一样的解码路径——生成的是结构化参数而不是自由文本。因此它的格式稳定性远高于提示词约束。缺点也很明显需要模型服务端支持函数调用能力且过程里多了一次模型先决定调用哪个工具、再生成参数的推理解码延迟略高一点。我的建议是这样的场景推荐方案字段少于 8 个、结构扁平、一次性任务PydanticOutputParser 或 StructuredOutputParser字段多、嵌套深、枚举值严格函数调用路线如 Instructor 封装需要兼容多种模型包括不支持函数调用的模型提示词约束路线对延迟极度敏感提示词约束路线函数调用延迟更高一个值得注意的趋势是新一代模型对 JSON 输出模式JSON Mode的支持越来越普遍不少模型服务商在 API 里直接就提供了强制输出合法 JSON的模式这种模式可以和 PydanticOutputParser 配合前者负责保证 JSON 解析合法性后者负责 JSON 到 Pydantic 的结构校验。我最近的几个项目都是这么搭配的解析失败率几乎降到了零。4. 链与解析器配合时的稳定化技巧只讲原理不讲稳定性等于白讲。链加解析器这套组合真正进入生产环境后才会暴露一堆问题。下面把我实际踩过的坑、验证过的兜底策略一条条列出来每条都来自真实线上事故或压测记录不是拍脑袋。4.1 异常处理的正确姿势识别三类典型失败第一类失败是格式错误模型输出不是合法 JSON。这最常见不管是json.loads抛异常还是 Pydantic 校验失败都属于这一类。兜底策略是捕获OutputParserException记录原始输出之后带着上一次缺少的字段重试一次。第二类失败是内容错误但格式正确JSON 解析成功字段都有但值是错的或答非所问。这种最容易被漏掉因为程序不会报错。我的兜底做法是加语义校验节点——比如要求输出里的component必须在预设组件白名单里否则返回无法识别并让上游重新处理。这种格式合法但语义错误的错误比格式错误更常见也更隐蔽。第三类失败是部分失败批量处理时一条数据出错导致整批回滚。应对思路是每一条数据独立 catch 异常让失败的记录单独进入重试队列不拖累其他正常数据。放一个我一直在用的鲁棒调用封装def safe_chain_invoke(chain, payload, retries2): for attempt in range(retries 1): try: return chain.invoke(payload) except Exception as e: if attempt retries: raise # 日志记录原始输入和异常内容便于事后复盘 print(f[重试] attempt{attempt}, error{str(e)[:100]}) return None4.2 从解析失败到自动修复增加一个格式修复节点单一try-except只能在不抛异常的情况下兜底。如果你的场景对成功率要求很高比如凌晨批量处理上万条工单哪怕 1% 的失败率也意味着上百条数据要人工接管这时候就需要自动修复。思路很简单失败后把原始输出和解析器的报错信息一起丢给一个专门负责改写 JSON的模型让它只负责修复格式、不改变内容fix_prompt PromptTemplate.from_template( 下面是一段大模型输出的内容未能通过 JSON 解析请修复为合法的 JSON只输出修复后的 JSON。\n 解析报错{error}\n原始输出{raw_output} ) def robust_parse(raw_output, error, llm): fixed llm.invoke(fix_prompt.format(errorerror, raw_outputraw_output)).content return parser.parse(fixed)自动修复节点的代价是额外一次模型调用但它比整条链重新跑一遍便宜得多——不需要重新处理前面的步骤。我实测在 3% 左右的初始失败率下一次自动修复能挽回其中 70%整体成功率能到 99% 以上。再配合重试队列基本可以做到批量任务无人值守。4.3 流式输出与解析器的兼容性问题流式输出是另一个高频踩坑点。你的链如果接了stream方法也就是逐字输出那么输出解析器的工作方式会发生变化——你不能等到全部输出结束后再解析那样流式就失去了意义。LangChain 提供了一套流式支持机制解析器可以对接transform()方法一边接收增量文本一边生成结构化结果。但实际经验告诉我LangChain 内置解析器对流式场景的支持并不理想。因为大多数解析器比如 Pydantic 解析器必须拿到完整 JSON 才能开始解析你没法在中途半句话的时候做结构化。所以如果你的场景是边输出边展示文本最终要交结构化数据我建议把两条链路分开展示链路用流式文本输出最终字段可信链路用非流式的完整解析。两边的模型调用可以并行用户看到的文本流不会卡住最后的解析结果也能拿到完整对象。另一个更高级的方案是使用流式专用解析器StreamingParser 这类第三方实现它会对增量文本做缓存、拼接、预判支持在输出过程中逐步填充 JSON 结构。但这种方案对模型的输出格式要求非常苛刻任何字段顺序变化都会导致流式解析混乱。我自己的建议是除非团队有余力维护流式解析的特殊兼容层否则默认使用展示与解析分离的方案稳定永远比炫技重要。5. 示例项目带路由和结构化输出的工单处理链絮絮叨叨讲了这么多原理和坑最后用一个完整的小项目把这些内容串起来。这里做一个工单处理链 Demo支持路由分类、结构化输出、失败重试全部代码都可以直接复制到本地跑。5.1 项目结构设计与完整代码项目目标是接收一条用户工单先判断类型bug / feature / question然后按照类型走不同提示词模板提取信息最终输出 Pydantic 对象。先定义数据模型from pydantic import BaseModel, Field from typing import Literal class TicketAnalysis(BaseModel): category: Literal[bug, feature, question] Field(description工单类型) module: str Field(description涉及模块) summary: str Field(description一句话概括) urgency: Literal[high, medium, low] Field(description紧急程度)然后定义三类不同的解析提示词。以 bug 类为例from langchain.output_parsers import PydanticOutputParser parser PydanticOutputParser(pydantic_objectTicketAnalysis) bug_prompt PromptTemplate.from_template( 你是运维工程师。分析这条 bug 工单提取结构化信息。\n工单{input}\n{format_instructions} ).partial(format_instructionsparser.get_format_instructions())路由逻辑上面已经提过这里不再重复。重点是把路由结果和解析链拼起来def route_ticket(info: dict): category llm.invoke(router_prompt.format(inputinfo[input])).content.strip().lower() selected_prompt {bug: bug_prompt, feature: feature_prompt, question: question_prompt}.get(category, question_prompt) return selected_prompt | llm | parser ticket_chain RunnableLambda(route_ticket) result ticket_chain.invoke({input: 用户反馈导出报表时一直转圈点击无响应怀疑是异步任务队列堵塞}) print(result.category) # bug print(result.urgency) # high 或由模型判断5.2 为什么用路由 结构化解析而不是一个全能链总有人问为什么不设计一个万能提示词一步直接完成分类和提取我试过效果不稳定。原因在于分类任务和提取任务对模型的注意力分配要求不同——分类时模型要抓全局语义提取时需要逐个字段扫描细节。把它们塞进一个提示词模型在长输出过程中经常漂移开头说得对结尾突然给字段加了怪东西。拆成路由加解析两步之后每一路模型的任务变得单一且明确。路由只需要输出一个词分类准确率显著提升各个分类下的解析提示词也各自精简对格式的遵循度更好。这就是工程里常说的单一职责——这个原则不仅适用于代码函数也适用于提示词设计。多提一句成本优化路由阶段用便宜的小模型解析阶段用能力强的大模型整体费用比全链路都用大模型低很多。我这边实际观测费用约为原来单一模型的 60%因为很多工单路由到 question 类question 类我用的小模型解析完全没压力。6. 实测参考参数、延迟与成功率观察这一节给出一些我在实际压测中观察到的参考数据。注意这些数据依赖具体模型版本和集群负载放出来不是为了让你直接抄参数而是为了让你建立哪些变量影响成败的敏感度。6.1 结构化输出延迟的构成拆解一次带结构化输出的链式调用延迟大致由四部分构成网络通信往返时间基础开销与链路无关。模型生成长度输出内容越长生成耗时越长。结构化输出一般是短输出所以这部分可控。路由判断耗时如果用了路由链先分类再进入子链多了一次模型调用延迟增加约 200-500ms。解析器解析耗时Pydantic 校验和 JSON 解析在毫秒级几乎可忽略。所以如果你觉得链路慢先看是不是多了一次路由调用再看生成的 token 数量。我见过有人把路由分类和内容生成放在同一个模型调用里就为了省那 200ms——但从稳定性和可维护性角度我认为不划算。6.2 不同解析方案的实测成功率对比我基于一个 500 条混合工单样本做过对比字段数量在 6 个左右含一个枚举字段方案一次成功率加自动修复后成功率纯提示词要求输出 JSON无解析器87%91%修复成本高PydanticOutputParsertemperature093%99%函数调用路线Instructor 封装97%99.5%可以看到提示词靠自觉的方案一次成功率只有 87%也就是说 100 条里有 13 条需要人工介入。纯提示词要求输出 JSON 是一个非常常见的初学者写法效果也确实最差——不是因为它不能用而是因为你把错误处理成本全留给了后续代码。表格里 PydanticOutputParser 的 93% 是多数人能轻松达到的水平加上自动修复节点后到 99% 也没问题。函数调用路线的 97% 一次成功率确实诱人但需要你的模型服务商支持且对延迟敏感服务未必友好。怎么选看你自己场景的容忍度。提示上面所有数据都来自我自己搭建的测试环境你可以把它们当作参考基准但不要当作行业标准。真正靠谱的做法是拿你自己的数据做一个几十条的小样本评估很快就能知道哪种方案在你的模型和数据分布下表现最好。
阅读完成 · 觉得有帮助？
