LLM Structured Output 生产工程:别再写正则解析JSON 了(工程师踩坑版)
你以为你在“接入大模型”,其实你在“接入一个不稳定的文本生成器”。
一旦你的业务链路里出现了:
JSON.parse(模型输出)—— 你就已经把事故种子埋进了生产。
我写这篇不是复述文档(文档告诉你“能用”,不会告诉你“会炸”)。
过去几个月我在做一个典型的 AI 工程系统:上游是业务请求,中间是 LLM 推理和若干工具调用,下游是严格依赖结构化数据的工作流(风控、工单、检索、路由、计费、监控)。
我们一开始也很朴素:让模型“返回 JSON”。效果在测试环境很好,上线之后开始出现一种非常阴险的故障:
- 99.9% 的请求都 OK
- 0.1% 的请求随机失败(带引号的名字、超长文本、用户复制了一段代码、模型被安全提示打断……)
- 失败之后触发重试
- 重试又失败 → 触发更大范围重试
- 队列堆积、线程打满、下游数据落库出现脏字段
这不是模型“变笨了”,而是你在用“文本协议”承载“数据协议”。
下面我用工程师视角把这事拆开:
- JSON mode / tool calling / structured output 到底差在哪
- 生产里会踩的 6 个坑(以及怎么把坑变成指标)
- 给你两套可直接复用的实现:TypeScript(Zod) / Python(Pydantic)
文章比较长(>5000字),但我保证:不是堆概念,是能直接搬进仓库的那种。
1. 三种“结构化输出”到底差在哪(别混为一谈)
很多团队把下面三件事统称“结构化输出”,然后就开始争论“到底选哪个”。
其实这三种完全不是一类东西:
1.1 Prompt 约定 JSON:最便宜,也最不可靠
典型写法:
- system:你是一个提取器
- user:请返回 JSON,字段有 a/b/c
优点:
- 便宜
- 实现简单
- 兼容任何模型
缺点:
- 没有协议保证:它“尽量”输出 JSON,不是“必须”输出 JSON
- 容易出现前后缀:
\njson\n{…}\n\n解释一段\n - 遇到边界字符(引号、换行、unicode)更容易破坏格式
我对这种方式的定位很明确:只能用于“失败代价极低”的场景。
1.2 Tool Calling / Function Calling:像 RPC,但不是强约束
很多平台支持把“输出 JSON”包装成一次函数调用:
- 你声明一个 tool schema
- 模型返回一个 tool call(携带参数)
- 你的程序拿参数当作结构化数据
这比纯 prompt 好,因为:
- 模型更愿意“遵守工具接口”
- 你可以在“工具层”做校验和默认值
但工程上要清醒:
- 这仍然可能出现字段缺失、值不在范围、enum 乱填
- 多轮对话时,模型可能在工具调用前后夹杂自然语言
- 模型可能调用了“你不希望它调用”的工具(需要 allowlist)
它的最佳用途是:需要动作编排(调用 API、查库、下单)而不是纯结构化抽取。
1.3 真正的 Structured Output:Schema-Constrained Decoding
这类能力的核心不是“提示词更强”,而是约束解码:
- 你给 JSON Schema(或等价的结构定义)
- 解码器在每个 token 步骤,把“不可能组成合法 JSON 的 token”直接屏蔽
- 结果是:输出在语法上满足 schema(结构正确、类型正确)
这类方案把“结构正确性”从应用层(regex/parse/retry)下沉到推理层,是质变。
但别误会:
- 它不保证语义正确(字段里填的内容仍可能胡说)
- 它对 schema 设计非常敏感(太深、太宽都会让质量下降)
结论:如果下游强依赖结构,structured output 是生产默认选项。
2. 生产级 Structured Output 的 6 个坑(我踩过的那种)
坑 1:流式输出被中断,截断 JSON 不是“坏运气”,是必然事件
很多同学喜欢 streaming,因为用户体验好、能做渐进渲染。
但对于 structured output,streaming 有一个天然矛盾:
- structured output 想要一个“完整闭合的 JSON”
- streaming 可能随时因超时、取消、网络抖动、上游限流而中断
于是你会拿到:
{"items":[{"id":1,"name":"a"},这时候你如果做JSON.parse(),必炸。
工程建议:
对“必须结构化”的链路,默认不要 streaming(或者只在 UI 层 streaming,而业务解析层拿完整结果)
如果必须 streaming,必须把协议变成:
- 流式输出自然语言(给用户看)
- 最后一段输出严格 JSON(给机器用)
并且要允许“最后 JSON 丢失”时回退到非流式重试。
坑 2:max_tokens / 超时导致 schema 破坏:要在协议层兜底
很多 structured output 的实现能保证“生成过程中合法”,但一旦你:
- 达到 max_tokens
- 触发超时
- 被上游取消
输出仍然会变成“半截”。
这不是模型问题,是传输层和资源控制层的问题。
建议:
- 把结构化输出的生成预算单独配置(不要和聊天混用)
- schema 要有“长度上限”策略(比如列表长度、字符串长度)
- 对关键字段,宁可短一点,也不要无限长
坑 3:optional / null 滥用:你以为在容错,其实在吞错
很多人为了“让它别报错”,把字段都做成 optional:
name?: stringscore?: number
这样 schema_valid_rate 可能很高,但你拿到的结构化对象里面全是null/缺失字段。
然后下游继续跑:
- 路由器拿不到 category → 默认走一个兜底模型 → 成本暴涨
- 风控拿不到 risk_level → 默认放行 → 真事故
建议:
- 关键字段不要 optional
- 真要 optional,也要在业务层做semantic reject(结构有效但语义无效)
坑 4:enum/范围约束:类型对了,值错了
即使 structured output 保证类型正确,也可能出现:
confidence: 5(明明约束 0~1)sentiment: "good"(明明 enum 是 positive/negative/neutral)
所以 schema 应该写“范围约束”和“枚举”,并且要把这些约束当成生产指标:
- 结构有效但违反范围 → 记为 semantic reject
- 触发 semantic reject → 进入修复流程(下一节代码会给你)
坑 5:嵌套太深:schema 不是越细越好
我见过最常见的“聪明反被聪明误”:
- 业务方把一个复杂的业务对象完整塞进 schema
- 嵌套 5~7 层
- 每层都有 optional/list/union
结果:
- 模型生成质量明显下降
- 结构虽然能闭合,但内容大量空洞
- 整体 token 成本飙升
经验值:
- 结构化输出适合 2~3 层嵌套
- 超过 3 层,建议拆成两段:先抽骨架,再补细节
坑 6:多模型/多供应商:同一 schema,稳定性差一个数量级
工程上你一定会做:
- 主模型 + fallback
- 不同任务路由不同模型
但 structured output 的稳定性并不均匀:
- 某些模型对 enum/范围支持好
- 某些模型对长列表支持差
- 某些模型一旦遇到复杂描述字段就开始啰嗦
建议:
- schema 也是“兼容性矩阵”的一部分
- 上线前做最小 eval:同一批样本跑 3 个模型,记录:
- schema_valid_rate
- repair_rate
- semantic_reject_rate
3. 端到端工程实现:TypeScript + Zod 的“两段式”管道
目标很简单:
让下游永远拿到一个typed object(或明确失败),而不是
string。
我推荐一个在生产里非常好用的模式:
- 硬约束阶段:尽可能一次得到 schema 合法对象
- 软修复阶段:如果失败,进入“修复模式”,让模型只做“修 JSON”,不做“重新理解任务”
下面这套代码可以直接拷走。
3.1 定义 schema:把“描述字符串”当成 prompt 的一部分
// structured_output.tsimport{z}from"zod";exportconstTicketTriageSchema=z.object({category:z.enum(["billing","bug","feature","security"]).describe("工单分类:计费/缺陷/需求/安全"),priority:z.enum(["p0","p1","p2","p3"]).describe("紧急程度:p0=线上故障, p1=严重影响, p2=一般问题, p3=咨询"),confidence:z.number().min(0).max(1).describe("模型对分类的置信度,0~1"),summary:z.string().min(10).max(200).describe("一句话摘要,10~200字"),actions:z.array(z.string().min(2).max(80)).min(1).max(6).describe("建议动作列表,最多6条"),});exporttypeTicketTriage=z.infer<typeofTicketTriageSchema>;注意:.describe()不是写给人看的,是写给模型看的。
经验:
- 描述要短、明确、带边界
- enum 的含义要写出来(否则模型会乱填)
3.2 调用 + 校验 + 指标:把失败变成可观测事件
下面写一个通用 client:
// llm_client.tsimport{z}from"zod";exporttypeLLMProvider={structured:<T>(args:{model:string;system:string;user:string;schema:z.ZodSchema<T>;timeoutMs:number;})=>Promise<unknown>;// 返回 raw};exporttypeMetrics={inc:(name:string,labels?:Record<string,string>)=>void;observe:(name:string,value:number,labels?:Record<string,string>)=>void;};exportasyncfunctionrunStructured<T>(args:{provider:LLMProvider;model:string;system:string;user:string;schema:z.ZodSchema<T>;timeoutMs:number;metrics:Metrics;}):Promise<{ok:true;value:T}|{ok:false;error:string;raw?:unknown}>{constt0=Date.now();try{constraw=awaitargs.provider.structured({model:args.model,system:args.system,user:args.user,schema:args.schema,timeoutMs:args.timeoutMs,});constparsed=args.schema.safeParse(raw);if(!parsed.success){args.metrics.inc("llm_schema_invalid_total",{model:args.model});return{ok:false,error:parsed.error.message,raw};}args.metrics.inc("llm_schema_valid_total",{model:args.model});args.metrics.observe("llm_latency_ms",Date.now()-t0,{model:args.model});return{ok:true,value:parsed.data};}catch(e:any){args.metrics.inc("llm_call_error_total",{model:args.model});return{ok:false,error:String(e?.message??e)};}}3.3 修复流程:不要“重试原任务”,要“修复 JSON”
很多人解析失败就“重试同样 prompt”。这会导致:
- 同样错误重复出现
- token 成本翻倍
- 失败时延更长
更稳定的做法:
- 让模型只做JSON 修复
- 输入是“失败的 raw 内容 + schema 要求”
// repair.tsimport{z}from"zod";exportasyncfunctionrepairToSchema<T>(args:{provider:{text:(a:{model:string;system:string;user:string;timeoutMs:number})=>Promise<string>;};model:string;schema:z.ZodSchema<T>;badText:string;timeoutMs:number;}):Promise<T>{constsystem="你是一个严格的JSON修复器。你只输出JSON本体,不要输出多余字符。";constuser=["把下面内容修复为满足给定schema的JSON。","要求:","1) 只输出JSON,不要markdown代码块,不要解释","2) 若字段缺失,请根据上下文合理补全;若无法补全,用最安全的默认值","3) 保持语义不变","\n[坏输出]",args.badText,].join("\n");constfixed=awaitargs.provider.text({model:args.model,system,user,timeoutMs:args.timeoutMs,});constobj=JSON.parse(fixed);constparsed=args.schema.parse(obj);returnparsed;}3.4 幂等与重试:避免“解析失败→重试雪崩”
建议:一次 structured + 一次 repair;仍失败就明确失败并降级异步处理。
4. Python + Pydantic:把业务规则写进 schema,而不是写进 prompt
fromtypingimportList,LiteralfrompydanticimportBaseModel,Field,field_validatorclassQueryIntent(BaseModel):intent:Literal["lookup","compare","troubleshoot","buy"]=Field(description="用户意图:查资料/对比/排障/购买")keywords:List[str]=Field(min_length=1,max_length=8,description="检索关键词列表")must_include:List[str]=Field(default_factory=list,max_length=5,description="必须包含的词")language:Literal["zh","en"]=Field(description="查询语言")@field_validator("keywords")@classmethoddefno_empty(cls,v:List[str]):v=[x.strip()forxinvifx.strip()]ifnotv:raiseValueError("keywords empty")returnv5. 观测与治理:把“输出质量”当成一等公民指标
三个核心指标:
schema_valid_raterepair_ratesemantic_reject_rate
6. 结尾:决策表 + 可复用模板
把“结构化输出”当成协议,而不是提示词技巧。
