企业官网建设流程全解析

输出在抖，不是模型坏了：LangChain 消息边界与结构化输出的稳定化改造

同一张工单，三次调用三种格式，问题并不在模型。

[读者] 把 LLM 接进业务系统的后端与平台工程师 [痛点] 输出时好时坏，解析靠正则兜底，重试越重越贵 [现在读] 这条链路已上线，抖动就是故障，不是体验问题 [读完] 能用消息边界加 Schema 契约把输出做成可交付接口

[旧方案] Prompt 里写只输出 JSON + StrOutputParser | v [新需求] 输出要直接进下游系统，字段可解析、可枚举、可空判定 | v [冲突] 自然语言输出没有契约，解析层只能靠猜 | v [后果] 偶发解析失败 + 重试放大成本 + 超时拖垮整批

我是老李，在一个做企业服务的团队里带平台组。我们有一套叫“工单裁决台”的系统，接客服工单的争议初判：客户投诉、退款争议、履约纠纷，先由模型给出一个初步判定，再决定是自动回复、转人工，还是升级到客诉组。

最初的版本很简单：一条 prompt，一次模型调用，把返回的文本丢进关键词匹配。那时候模型的输出只给人看，人瞟一眼就知道该点哪个按钮，格式丑一点无所谓。

后来业务要求把那一下点击也省掉——判定结果直接写回工单系统的字段里。同一份 prompt 原封不动搬过去，第二天我们就开始收到解析失败的告警。

你有没有想过，当输出从“给人看的文字”变成“给系统吃的字段”，你还在用 json.loads 加 try/except 兜底吗？

同一份 prompt，为什么在测试环境稳如老狗，到了生产会漂成三种格式？

如果重试会放大成本、超时又会拖垮整批，可用性的最后一道坎，到底该由谁来兜？

01、故事

工单裁决台上线的第三个月，值班群里开始出现一种很规律的告警：每天 21 点到 23 点，工单积压量翻倍。白天它跑得很好，晚上就崩。

我们给自己定的任务是：把争议工单的初判从人工 3 分钟压到模型几秒钟，并且判定结果要能直接驱动下游动作——自动回复、转人工、升级客诉组。人不再看模型说了什么，只看系统据此做了什么。

转折点是“换模型”。因为成本，我们把后端换成了 OpenAI 兼容网关后面的一批模型，随时可切。切的那天我们发现：同一张工单，A 模型返回一段带前言的自然语言，B 模型返回被 Markdown 围栏包住的 JSON，C 模型返回规范 JSON 却把布尔值写成中文的“否”。三种脾气，共用一套解析代码。

于是链路里出现了一层层补丁：先 strip，再找第一个左花括号，再找最后一个右花括号，再用正则把“是/否”映射成布尔值，最后再拿字典的 get 挨个取字段。补丁写完那周，解析确实不报错了——直到某个晚上，一个模型把所有字段名改成了驼峰式。

我们开始怀疑模型“不听话”。这个怀疑方向，后来被证明只对了一半。

格式先崩
语义后崩
补丁再多
也只遮眼

02、问题

旧方案的本质是一句话：把自然语言输出，当成结构化输出来解析。StrOutputParser 做的事情是把模型返回的消息对象里的文本内容拼成字符串，它不校验、不约束、不感知你的字段定义。约束只存在于 prompt 里那句“请只输出 JSON”，而 prompt 是建议，不是契约。

业务影响很直接。解析失败的那次调用，工单会掉进一个空的 catch 分支：既不自动回复，也不转人工，它卡在那儿，直到积压告警。更糟的是“解析成功但语义错”的情况——模型返回了合法 JSON，但 need_escalate 是字符串“否”，下游按真值判断，一个字符串在 if 里恒为真，所有该转人工的工单全被自动回复了。

技术表现可以归成四类：一是格式违约，返回里有前言、有 Markdown 围栏、有多余说明；二是字段漂移，字段名、大小写、枚举取值在模型之间不一致；三是长度失控，为了“让模型判断更准”，我们把整段工单历史塞进上下文，token 随对话轮次线性增长；四是时间无界，调用没有显式超时，SDK 层的默认重试叠加业务层的 while 重试，一个慢请求可以拖住一整批串行任务。

所以我们把“修好”定义成五条可验证的完成标准，而不是“看起来稳了”：

• 对同一批固定评测样本，解析成功率由脚本跑出来，不靠体感；
• 每个字段的类型与取值域在代码里静态声明，不依赖事后正则修补；
• 单次调用有明确的时间上限，最坏耗时可计算；
• 重试有次数预算，且预算耗尽后必然进入确定的降级路径，不向业务层抛异常；
• 单请求 token 有上限，超限走同一条降级路径。

注意这五条里没有一条是“让模型更聪明”。这不是偶然。

不是模型错
是契约缺席
标准先立住
才谈得上稳

03、原理

消息角色不是装饰，是边界。LangChain 的消息体系把一次调用拆成 System、Human、AI、Tool 四类：System 承载稳定不变的行为约定，Human 承载每次变化的数据，AI 承载历史回复，Tool 承载外部工具返回的事实。把“只输出 JSON”写在 Human 里，它会随着每张工单的正文一起被稀释；写在 System 里，它每次都在同一个位置、以同一种权重出现。而 ToolMessage 的意义在于：外部事实必须标明来源，不能让模型自己编一段塞进 AI 消息里冒充事实。

上下文是被计费的边界。每多一条消息，就多一份 token，也多一份让模型“参考”的东西。工单系统里最常见的错误是把整条工单历史全量拼进上下文，理由是“信息越全越准”。实际上历史越长，模型越容易把上一轮的输出格式带到这一轮——这也是为什么同一份 prompt 在测试环境稳定、在生产漂移：测试用例是三行，生产输入是三百行。

结构化输出的关键在输出侧，不在提示侧。这是本篇第一个反直觉判断：把约束从 prompt 搬到 schema，比在 prompt 里写十遍“务必输出 JSON”有效得多。原因在于位置不同——prompt 是对生成过程的建议，schema 是对解码结果的约束（在支持 json_schema 一类能力的后端上）或在客户端执行的解析加校验（在不支持的后端上）。用 Pydantic 定义模型，再用 with_structured_output 绑定，本质是把“格式对不对”从运行时赌博变成了调用前就写死的合同。具体能力矩阵与参数组合，需按当前官方文档核验。

第二个反直觉判断：温度调到 0 不等于输出确定。它降低的是采样随机性，不改变格式违约的概率结构，也不消除后端批处理、浮点累加顺序带来的差异。如果你的失败样本是“返回里多了一段前言”，那降温度只会让模型稳定地继续多那段前言。

第三个反直觉判断：重试不是稳定性手段，是有预算的补偿。稳定性来自“失败有确定的归宿”，而不是“失败多试几次”。没有预算的重试，只是把一次失败推到更贵、更晚的地方。

边界先划清
约束放输出
温度非确定
重试需有价

04、架构

[输入] 工单标题 + 正文 + 历史摘要 | v [模块] System 契约 + Human 数据 + ChatPromptTemplate 管道 | v [数据/状态] Pydantic Schema（字段类型、枚举、描述） | v [处理] 结构化输出绑定 + 超时 + 预算内重试 + 降级 | v [输出] Judgement 对象 / 降级对象（同一类型）

这条链路里，真正起作用的不是那个管道符号，而是三个边界。第一个是输入边界：进入 Human 的只有标题、正文和一段被裁剪过的历史摘要，不是全量历史；第二个是输出边界：模型返回的东西必须先变成一个 Pydantic 实例，才算调用成功；第三个是时间与成本边界：timeout、max_retries 和 max_tokens 三个旋钮，决定了这次调用最坏能花掉多少秒、多少钱。

收益是可解析性从“运行时靠猜”变成了“调用前就写死”。下游拿到的一定是 Judgement 对象，字段类型固定，枚举值可枚举，不需要再写正则。降级对象和正常对象是同一个类型，下游不需要写两套分支。

代价也是真的。第一，Schema 变成了一份要维护的契约：改字段要同步下游，改枚举要考虑旧数据的兼容。第二，strict 一类的严格模式并不是所有模型、所有兼容网关都支持，换后端时要重新验证，否则会静默退回到“提示侧约束”。第三，结构化输出本身有额外开销——schema 要作为约束传给模型，token 会多一些。

适用条件是明确的：输出要进系统、字段有清晰取值域、业务能接受一次降级。如果输出只是给人看的一段解释文案，这套东西就是过度设计。

入口要收窄
出口要定型
时间要封顶
三界先立稳

05、实战一次

环境与依赖：Python 3.11 以上，安装 langchain、langchain-openai、pydantic。本节未在当前环境实测，以下为预期结果。

pipinstall-Ulangchain langchain-openai pydantic

核心实现是一个最小但完整的 V1：用 ChatPromptTemplate 组装 System 与 Human，用 ChatOpenAI 指向 OpenAI 兼容网关，用 StrOutputParser 拿字符串，再手工 json.loads。这个版本会跑通，也会把问题留在明面上。

# v1_pipeline.pyimportjsonfromlangchain_core.promptsimportChatPromptTemplatefromlangchain_core.output_parsersimportStrOutputParserfromlangchain_openaiimportChatOpenAI SYSTEM='你是工单争议判定助手。请只输出 JSON，字段为 need_escalate、reason、action。'HUMAN='''工单标题：{title} 工单正文：{content}'''prompt=ChatPromptTemplate.from_messages([('system',SYSTEM),('human',HUMAN),])model=ChatOpenAI(model='your-model-name',base_url='https://your-gateway.example.com/v1',api_key='sk-your-key',temperature=0.7,)chain=prompt|model|StrOutputParser()defjudge(title:str,content:str)->str:returnchain.invoke({'title':title,'content':content})if__name__=='__main__':raw=judge('退款被拒','客户称商品已签收超过七天，要求全额退款。')print(raw)print(json.loads(raw))

启动与验证：直接 python v1_pipeline.py。示例输出（预期）如下——第一次打印出的是模型原始文本，第二次解析在多数情况下会抛出 JSONDecodeError，报错停在第一行第一个字符，因为字符串开头是一句“好的，下面是判定结果”。

# 示例输出（未在当前环境实测，以下为预期结果）# 第一次打印：一句中文前言 + 被 Markdown 围栏包住的 JSON 片段# 第二次打印：json.decoder.JSONDecodeError: Expecting value: line 1 column 1 (char 0)

V1 跑通的意义在于：它证明了模型确实能做这件事，判断内容基本可用；同时它把三个问题原样留在了代码里——解析没有契约、调用没有超时、失败没有归宿。这三条，就是后面两章要逐个拆掉的东西。

先让它能跑
再让它露怯
问题留在明处
才好逐个拆

06、排查

现象一：解析失败，异常是 JSONDecodeError，集中在夜间批量任务。怀疑是模型不听话、随机性太大。检查动作是把原始返回体落盘，而不是只看异常栈。证据很清楚：失败的样本里，返回字符串的开头是一句中文前言，JSON 被 Markdown 围栏包着；成功的样本里，模型偶尔也会带一句总结。根因不是随机性，是 StrOutputParser 只做字符串拼接，链路上根本没有任何 schema 约束，模型每一次都在按自己的训练分布自由发挥。修复方向是结构化输出，而不是继续加固正则。

错误尝试：我们先把 temperature 从 0.7 降到 0，上线观察两天，认为“这样模型就不会乱来了”。为什么错：失败样本是格式违约，不是采样随机性。温度 0 只是让模型稳定地继续违约——它让失败变得更可复现，而不是更少。真正解决问题的是后来把约束加到输出侧。

现象二：解析成功但业务出错，need_escalate 被写成中文“否”，下游按真值判断全部走成了“需要升级”，转人工队列瞬间打满。怀疑是模型中文理解有问题。检查动作是把 Prompt 和返回体对照看。证据是：Prompt 里从头到尾没告诉模型这个字段是布尔值，更没给过取值域。根因是字段缺少取值域契约，模型只能猜一个它觉得自然的值。修复办法是在 Pydantic 里用 bool 类型加 Field 描述，枚举字段用 Literal 把取值域钉死。

现象三：整批任务偶发卡住，单张工单正常，批量就超时。怀疑是模型服务慢。检查动作是看耗时分布和异常类型，而不是看平均值。证据是：绝大多数请求很快，极少数请求一直没有返回，而业务层的重试逻辑是在外层套了一圈 while，SDK 内部还有自己的默认重试，两层叠加之后，一个慢请求能占住整批的串行进度。根因是调用没有显式超时，重试也没有预算。修复方向是把 timeout 显式设上，把 max_retries 关掉，重试收敛到业务层并带上次数预算，预算耗尽就降级。

三条链路指向同一个结论：这三类故障里，没有一类是靠“换个更强的模型”解决的。

先看原始体
再对字段义
最后量长尾
三类不同因

07、优化

第一处修改，针对格式违约。把 Pydantic 模型定义成输出契约，字段类型、取值域、描述都写进去，然后用 with_structured_output 绑定到模型上。原因是这层约束发生在解析侧，模型返回后立刻被校验成对象，不合法就直接失败，不会流到下游。新行为是：调用成功即得到 Judgement 实例，不再有“字符串里找花括号”这一步。

# v2_structured.pyfromtypingimportLiteralfrompydanticimportBaseModel,Fieldfromlangchain_openaiimportChatOpenAIclassJudgement(BaseModel):need_escalate:bool=Field(description='是否需要人工升级')severity:Literal['low','normal','high']=Field(description='争议严重度')reason:str=Field(description='不超过 60 字的判定依据，必须引用工单中的事实')action:str=Field(description='下一步动作，必须是一个可执行指令')model=ChatOpenAI(model='your-model-name',base_url='https://your-gateway.example.com/v1',api_key='sk-your-key',temperature=0,timeout=20,max_retries=0,max_tokens=400,)structured=model.with_structured_output(Judgement,method='json_schema',strict=True)

这里有两个参数选择值得说明。一是 max_retries 设成 0：不是不要重试，而是不许 SDK 偷偷重试，重试必须发生在业务层，因为只有业务层知道预算和降级路径。二是 timeout 设成显式值：让最坏耗时可计算，这是完成标准里写死的一条。strict 与 method 的具体取值组合，需按当前官方文档与所用兼容网关逐项核验，不是所有后端都支持。

第二处修改，针对没有归宿的失败。把重试、降级、超时收敛到一个函数里，并且让降级返回的对象和正常返回的对象是同一个类型。

# v2_fallback.pyfromlangchain_core.promptsimportChatPromptTemplatefromv2_structuredimportJudgement,structured SYSTEM='你是工单争议判定助手。只依据给出的工单内容判定，不要引入外部信息。'HUMAN='''工单标题：{title} 工单正文：{content}'''prompt=ChatPromptTemplate.from_messages([('system',SYSTEM),('human',HUMAN),])chain=prompt|structureddefjudge_safe(title:str,content:str,budget:int=2)->Judgement:last_err=Nonefor_inrange(budget+1):try:returnchain.invoke({'title':title,'content':content})exceptExceptionase:last_err=ereturnJudgement(need_escalate=True,severity='normal',reason='自动判定不可用，转人工：'+type(last_err).__name__,action='push_to_human_queue',)

新行为是：无论是解析失败、超时还是限流，业务层拿到的永远是一个 Judgement，最坏情况是 need_escalate 为真、action 指向人工队列。验证方式是构造三类输入——正常工单、超长工单、故意填错 base_url 的工单——观察前者返回结构化结果、后两者落到同一条降级路径且不抛异常。这里给出的是验证方法，不宣称任何成功率数字。

第三处修改，针对上下文与成本。把历史压缩成一段摘要在 Human 里给出，而不是把全量对话都塞进去；max_tokens 设上限，超限同样走降级。原因是上下文每扩一段，token 成本、格式漂移概率和延迟是同时上升的。

约束放前端
归宿留在后
预算写进码
三改各有由

08、演进

[同一输入：争议工单] | +--[V1] 提示侧约束 + 字符串解析 / 失败即抛异常 / 无超时无预算 | +--[V2] Schema 约束 + 显式超时 + 预算内重试 + 同类型降级 | [Trade-off] 得到：可解析、可枚举、可预期的失败归宿 失去：字段扩展的自由度，以及一部分后端的可移植性 边界：Schema 变更需同步下游，strict 能力需按后端核验

正确性上，V1 的正确性取决于运气——同一份输入可能在解析这一步就丢了；V2 把正确性拆成两层：格式正确由 schema 保证，语义正确交给评测集，职责不再混在一起。

稳定性上，V1 的失败是可扩散的，一次解析失败能拖住一整批；V2 的失败是可收敛的，最坏情况是一张工单进人工队列，不会向上冒泡。这是两代方案最本质的差别：不是 V2 失败更少，而是 V2 的失败有边界。

复杂度上，V2 确实更高：多了一个 Schema 要维护、多了一段降级逻辑、多了一份和后端能力的适配工作。这份复杂度换来的是可观测性——失败可以被分类计数，而不是笼统地记一个“报错”。

成本上，V2 多了 schema 带来的 token 开销，但省掉了失败重试和补丁维护。是否划算，取决于失败率，这个要自己测。

适用范围上，V1 适合原型验证和输出只给人看的场景；V2 适合输出要进系统的场景。这不是先进与落后，是场景不同。

遗留问题也很明确：多模型切换时，strict 能力不一致；结构化降级会让人工队列承担兜底压力，需要容量预案；Schema 演进缺少版本化机制。

旧版赌运气
新版留退路
稳不是不错
是错得有界

09、洞见

9.1 输出的确定性来自契约，不来自参数

我们最初的所有排查动作，都落在“怎么让模型更听话”上：改措辞、加感叹号、降温度、换模型。这些都收效甚微，因为它们在调整的是建议的强度，不是约束的强度。反直觉判断：把“只输出 JSON”从 prompt 搬到 schema，效果远大于把 temperature 从 0.7 调到 0。前者改变的是失败会不会流到下游，后者只改变失败发生的频率。

9.2 消息角色划分的是责任，不是格式

System、Human、AI、Tool 四类消息，最初我以为只是给模型看的标签。后来才明白它划分的是责任边界：System 是你对行为负责的部分，Human 是用户负责的部分，AI 是历史，Tool 是外部事实负责的部分。把业务约束写进 Human，等于让它和用户输入一起被稀释；把模型自己编的内容当 AI 消息塞进历史，等于让幻觉获得“事实”的身份。上下文边界管理不好，成本和正确性会同时失守。

9.3 重试是补偿，不是稳定器

反直觉判断：重试次数增加，系统可用性未必上升，有时反而下降。原因是没有预算的重试会把失败放大成雪崩——第一次超时占住一个连接，第二次重试再占一个，第三次把队列填满。真正的稳定性来自“失败有确定归宿”，也就是降级路径存在且被验证过。降级不是认输，它是可用性的下限本身。

9.4 换模型不是换参数，是换了一份隐性契约

我们一直以为接的是 OpenAI 兼容服务，所以换模型等于换一个 model 名字。实际上兼容的是接口形状，不是输出行为。不同模型对“只输出 JSON”的理解、对枚举值的默认表达、对长度的偏好，全都不一样。所谓兼容层，兼容的是协议，不是脾气。这也解释了为什么同一份 prompt 在测试环境稳定、在生产漂移——测试环境只跑过一个模型。

契约胜参数
责任分角色
重试需有价
兼容非同脾

10、系统落地

原来我们有的，是一条 prompt 加一个 StrOutputParser，加上一堆事后补丁：strip、找花括号、正则映射枚举、外层 while 重试。它能跑，但它的正确性依赖输入长得像测试用例。

本篇新增了三样东西。第一是输出契约：一份 Pydantic 模型，字段类型、枚举取值、字段描述都写死，配合结构化输出绑定，让“格式对不对”在调用返回那一刻就有结论。第二是失败归宿：一个带预算的重试循环，预算耗尽必然产出同类型的降级对象，业务层不需要写第二套分支。第三是资源边界：显式 timeout、显式 max_tokens、显式关闭 SDK 内部重试，把最坏耗时和最坏成本都变成可计算的值。

现在能做的，是把工单初判真正做成一个接口，而不是一段文本。下游拿到的永远是 Judgement，格式问题不再需要下游兜底；人工队列只处理“自动判定不可用”这一种确定的输入，而不是随机涌入的各种残片。

还缺什么？缺评测。现在我们知道格式稳定了，但不知道判断准不准——这两件事必须分开度量，格式靠 schema，语义靠固定评测集。也缺观测：失败目前只记了类型名，没有按模型、按网关、按工单类型拆分，无法定位是哪个后端的脾气变了。还缺 Schema 的版本管理，字段一旦变更，历史数据怎么兼容还没有答案。

下一步的演进方向是清楚的：把 Schema 变成有版本号的资产，把降级率和失败分类接进监控，把多后端的能力差异做成一张需要逐项核验的兼容表。这三件事做完，这套链路才算真的能长期维护。

有契约可依
有退路可走
有边界可算
才算落了地

11、小结

Q1 → 输出进系统就别再靠正则兜底：用 Pydantic 定义字段类型与取值域，用结构化输出绑定，让格式在返回那一刻定案。 Q2 → 漂移不是因为不稳定，而是因为约束只写在 prompt 里，且测试环境只跑过一个模型；约束搬到输出侧，换后端才可控。 Q3 → 由降级路径来兜：显式 timeout + 预算内重试 + 同类型的失败对象，让最坏情况可计算、可承接。 状态 → 一条可解析、可降级、可计费的 LLM 调用链路，V1 跑通，V2 立住边界。

一问格式稳
二问漂移因
三问谁兜底
答在契约中

12、作业

12.1 理解题：为什么把“只输出 JSON”从 prompt 搬到 schema，效果比降温度更明显？

参考答案：两者作用的位置不同。prompt 是对生成过程的建议，模型可以在遵循建议的同时加上前言、围栏或改写字段名；schema 作用在结果校验或解码约束上，不合法就是失败，失败会走确定的处理路径。降温度只降低采样随机性，不改变格式违约的概率结构，甚至会让违约变得稳定复现。前者改变的是“失败会不会流到下游”，后者改变的是“失败发生的频率”。

12.2 实战题：给 Judgement 增加一个 evidence 字段，要求必须引用工单中的原文片段，并保证旧代码不崩。

参考答案：在 Pydantic 模型里加一个带默认值的字段，例如 evidence: str = Field(default=‘’, description=‘引用工单中的原文片段，不超过 80 字’)。给默认值是为了让降级路径和旧调用方仍能构造出对象；结构化输出绑定不需要改写，schema 会自动带上新字段。同时要给评测集补一条校验：evidence 为空时是否可接受，需要业务明确。字段一旦成为契约，就要考虑旧数据兼容。

12.3 排障题：上线后解析失败率归零，但转人工队列反而变长了，怎么查？

参考答案：这说明失败从“抛异常”变成了“走降级”，故障没有消失，只是被隐藏了。排查顺序是：先按失败类型给降级路径打点分类，区分是 schema 校验失败、超时还是限流；再看是否集中在某个模型或某个网关上，这一步能验证“换后端换脾气”的假设；然后核对 timeout 与 max_retries 的实际取值是否生效，特别是 SDK 内部重试是否被真的关掉；最后检查 max_tokens 是否过小导致输出被截断，截断在高频场景下会稳定触发校验失败。关键是要让降级可计数、可分类，否则它就是个黑洞。

12.4 架构判断题：团队提出把重试次数从 2 次调到 5 次来提升成功率，你同意吗？

参考答案：不建议直接调。重试次数不是独立旋钮，它和 timeout、并发度、下游容量是一个整体。在超时场景下，重试会成倍占用连接和时间预算，把单点慢请求放大成批处理雪崩；在限流场景下，重试只会加剧限流。正确做法是先按失败类型拆分，对可重试的瞬时错误给有限预算，对不可重试的格式错误直接降级，并且让降级路径本身被监控验证过。判断标准不是“成功率涨没涨”，而是“最坏耗时可不可计算、失败有没有确定归宿”。

题问位置差
题问契约变
题问黑洞处
题问预算先

13、思考

这篇的核心冲突只有一句话：当输出从给人看变成给系统吃，稳定的定义就变了。给人看的输出，稳定意味着“读起来合理”；给系统吃的输出，稳定意味着“出错时有确定的归宿”。这两种稳定，需要的工程手段完全不同。

我们花了很多时间在调提示词和调温度上，那是在优化第一种稳定。真正把问题解决的，是三件看起来很朴素的事：把约束放到输出侧，把失败收敛到一条路径，把时间和成本写成显式的数字。它们都不聪明，但它们可验证。

换模型这件事也值得再想一层。所谓 OpenAI 兼容，兼容的是接口形状，不是行为脾气。任何一次后端切换，本质上都是一次隐性契约的变更，必须重新验证，而不是改个名字就上线。

最后留一句给自己的判断标准：判断一条 LLM 链路是否真的稳，不看它成功时多漂亮，看它失败时去哪里。

稳在失败处
不在成功时
契约先立住
脾气任它变