看到"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_input=user_input)) result_b = llm.invoke(prompt_b.format(context=result_a)) result_c = llm.invoke(prompt_c.format(data=result_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(input=info["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_object=IssueReport) prompt = PromptTemplate( template="请分析下面的技术工单，输出结构化信息。\n工单内容：{input}\n{format_instructions}", input_variables=["input"], partial_variables={"format_instructions": parser.get_format_instructions()}, ) llm = ChatOpenAI(model="gpt-4o-mini", temperature=0) 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, retries=2): 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(error=error, raw_output=raw_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_object=TicketAnalysis) bug_prompt = PromptTemplate.from_template( "你是运维工程师。分析这条 bug 工单，提取结构化信息。\n工单：{input}\n{format_instructions}" ).partial(format_instructions=parser.get_format_instructions())路由逻辑上面已经提过，这里不再重复。重点是把路由结果和解析链拼起来：
def route_ticket(info: dict): category = llm.invoke(router_prompt.format(input=info["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%（修复成本高） |
| PydanticOutputParser（temperature=0） | 93% | 99% |
| 函数调用路线（Instructor 封装） | 97% | 99.5% |
可以看到，"提示词靠自觉"的方案一次成功率只有 87%，也就是说 100 条里有 13 条需要人工介入。纯提示词要求输出 JSON 是一个非常常见的初学者写法，效果也确实最差——不是因为它不能用，而是因为你把错误处理成本全留给了后续代码。
表格里 PydanticOutputParser 的 93% 是多数人能轻松达到的水平，加上自动修复节点后到 99% 也没问题。函数调用路线的 97% 一次成功率确实诱人，但需要你的模型服务商支持，且对延迟敏感服务未必友好。怎么选，看你自己场景的容忍度。
提示：上面所有数据都来自我自己搭建的测试环境，你可以把它们当作参考基准，但不要当作行业标准。真正靠谱的做法是拿你自己的数据做一个几十条的小样本评估，很快就能知道哪种方案在你的模型和数据分布下表现最好。
