一、引言:一个“看起来很简单”的工程问题
凡是把 LLM 接进过生产系统的团队,几乎都被同一件事折磨过:模型输出不是合法 JSON。
典型症状包括但不限于:
- 输出被
```json代码块包裹,或者前面加一句“好的,以下是解析结果:”; - 字段缺失、字段名拼错(
userNamevsuser_name); - 类型漂移:该是数字的返回了
"12 岁",该是数组的返回了逗号分隔的字符串; - 枚举幻觉:约定的
status只有pending/success/failed,模型返回了processing; - 长输出被
max_tokens截断,JSON 在半个字符串里戛然而止。
很多人的第一反应是“在 prompt 里再强调一遍”。这在 demo 阶段有效,一旦进入线上,面对长上下文、多轮对话、并发调用,失败率就会稳定地停在 3%~15% 这个区间——对于日均十万次调用的系统,这意味着每天都在产生数千条脏数据。
更具体地说,一个电商订单解析场景中,模型需要从用户手写的自然语言里抽出商品名、数量、规格、收货地址。如果没有结构化输出保障,你会遇到“三件”而不是3,“朝阳区”被塞进city字段而district为空,甚至整个 JSON 被一句“好的,我帮您整理如下:”挡在前面。下游的库存系统、风控系统、工单系统一旦拿到这种数据,轻则报错重试,重则写入错误状态,需要人工清洗。
本文的核心观点是:结构化输出不是 prompt 问题,而是工程问题。可靠的做法是三层防线——语法层约束解码、语义层 Schema 校验、失败层错误回填重试。下面用 Python 把这套东西完整实现一遍。
二、核心原理:从“求它”到“约束它”
2.1 约束强度的三个层级
第一层:Prompt 约束。在 system prompt 里写“你必须只输出 JSON,不要任何解释”。它依赖模型的指令遵循能力,是概率性的,不可靠。模型在长上下文里会“忘记”格式要求,在多轮对话里会被用户的最新指令带偏,在采样温度较高时更容易自由发挥。你可以把它理解为“口头约定”,适合做第一道提醒,但不能作为生产系统的唯一保障。
第二层:JSON Mode。OpenAI 的response_format={"type": "json_object"}、多数国产模型对应的json_object参数。它保证输出是语法合法的 JSON,但不保证符合你的字段结构——模型完全可能返回{"result": "..."}。JSON Mode 的底层实现通常是在解码阶段限制 token 集合,使得任意时刻生成的内容都符合 JSON 语法。但它不关心你的业务 Schema,所以字段名、字段类型、枚举值仍然可能出错。
第三层:Schema 约束(Structured Outputs / 受约束解码)。OpenAI 在 2024 年 8 月推出的 Structured Outputs,以及开源侧的outlines、XGrammar、llguidance,走的是另一条路:把 JSON Schema 编译成有限状态机(FSM),在每一步解码时,把所有会让 FSM 进入死状态的 token 的 logit 直接置为-inf。模型在物理上不可能输出非法 token。举个例子,如果你的 Schema 要求priority只能是P0/P1/P2/P3,当模型解码到该字段的值时,FSM 只允许能拼出这四个字符串的 token 被采样。模型即使“想”输出processing,也会在第一步就被屏蔽掉。
这是目前最强的保证——但仍有边界:它只保证结构合法,不保证语义正确。模型完全可以填一个语法无误但业务上错误的枚举值,比如把“非常紧急”的工单标成P3。因此,约束解码解决的是“格式合规”,业务正确性仍需要校验、规则引擎或人工复核。
2.2 校验是契约,不是事后检查
别用json.loads()之后手写if判断字段。用 Pydantic 把“期望的输出结构”声明成类型,让校验逻辑和数据结构同源:
fromtypingimportLiteralfrompydanticimportBaseModel,FieldclassTicket(BaseModel):title:str=Field(min_length=1,max_length=80)priority:Literal["P0","P1","P2","P3"]tags:list[str]=Field(default_factory=list,max_length=5)need_human:boolLiteral把枚举约束写进 Schema,Pydantic 在运行时做二次兜底,model_json_schema()还能直接喂给模型做约束解码。一份定义,三处复用。
这里的关键思想是“契约先行”:你先把期望的输出结构定义清楚,然后让模型、校验器、下游系统都围绕这份契约工作。model_json_schema()生成的 JSON Schema 可以直接作为 API 的response_format参数,也可以存进配置中心,供多个服务共享。Pydantic v2 的model_validate_json()会一次性完成 JSON 解析和类型校验,比json.loads()+ 手动判断快得多,也安全得多。
此外,Field还能表达更多约束:min_length、max_length、pattern、ge、le等。例如,如果你希望title不能为空,priority必须是四个枚举之一,tags最多五个,这些约束都会体现在 Schema 里。如果模型使用支持 Structured Outputs 的接口,这些约束会在解码阶段生效;如果模型不支持,它们会在校验阶段被 Pydantic 捕获,并进入重试流程。
2.3 重试的本质:把错误变成上下文
校验失败时,不要盲目重发同一个请求。正确的做法是把校验错误信息结构化地回填给模型,让它知道哪里错了:
你上一次的输出校验失败:
priority: 值processing不在允许的枚举['P0','P1','P2','P3']中tags: 期望数组,实际为字符串
请只输出修正后的 JSON,不要任何解释。
这个“错误回填循环”通常能把二次成功率拉到 90% 以上。
为什么有效?因为大模型具备上下文学习能力。你把它刚才生成的错误输出作为assistant消息放回对话,再以user身份给出具体的校验错误,模型会把这些错误当作“待修正的反馈”,而不是重新生成一个无关的回答。相比单纯重复“请输出 JSON”,错误回填提供了精确的修正方向:哪个字段、什么类型、期望什么值。这就像代码编译器告诉你行号和错误类型,而不是只说“编译失败”。
但要注意,错误回填不是万能的。如果模型本身不具备理解 Schema 的能力,或者上下文过长导致错误信息被淹没,重试仍然可能失败。因此,重试次数要有上限,通常 2~3 次足够。超过之后应该降级到人工处理、默认值或备用模型,而不是无限循环。
三、实战:一个可落地的结构化输出网关
3.1 脏文本 JSON 提取器
即使开了 JSON Mode,模型仍可能加前后缀。先写一个健壮的提取器,优先取代码块,否则用括号配对扫描(注意跳过字符串内的括号和转义符):
importre,jsonfromtypingimportOptional FENCE_RE=re.compile(r"```(?:json)?\s*(.*?)```",re.S|re.I)PAIRS={"{":"}","[":"]"}def_scan_balanced(s:str)->Optional[str]:"""从 s 中截取第一个括号配对完整的 JSON 片段"""start,stack=None,[]in_str,esc=False,Falsefori,chinenumerate(s):ifin_str:ifesc:esc=Falseelifch=="\\":esc=Trueelifch=='"':in_str=Falsecontinueifch=='"':in_str=TrueelifchinPAIRS:ifstartisNone:start=i stack.append(ch)elifchin"}]":ifstackandPAIRS[stack[-1]]==ch:stack.pop()ifnotstack:# 完整闭合,直接返回returns[start:i+1]else:# 括号错配,重置start,stack=None,[]returnNonedefextract_json(text:str)->Optional[str]:m=FENCE_RE.search(text)ifm:candidate=_scan_balanced(m.group(1))orm.group(1).strip()ifcandidate:returncandidatereturn_scan_balanced(text)这个提取器的核心是一个小型状态机。它逐字符扫描,维护三个状态:是否在字符串内、是否处于转义、括号栈。为什么不能只用正则?因为 JSON 允许嵌套对象和数组,正则很难正确处理嵌套深度,也容易被字符串里的{、}干扰。括号配对扫描可以保证截取到第一个完整闭合的 JSON 片段,即使模型在前面写了“好的,以下是结果:”,在后面写了“希望有帮助”,也能准确提取。
测试用例可以帮助你验证提取器的健壮性:
assertextract_json('```json\n{"a": 1}\n```')=='{"a": 1}'assertextract_json('好的:{"a": {"b": 2}} 完毕')=='{"a": {"b": 2}}'assertextract_json('字符串里的括号 {"a": "} not end"}')=='{"a": "} not end"}'assertextract_json('没有 JSON')isNone3.2 校验 + 回填重试的主循环
frompydanticimportBaseModel,ValidationErrorclassStructuredOutputError(RuntimeError):passdef_repair_prompt(err:ValidationError,schema:dict,limit:int=5)->str:lines=[]foreinerr.errors()[:limit]:loc=".".join(str(x)forxine["loc"])or"<root>"lines.append(f"- `{loc}`:{e['msg']}(type={e['type']})")return("上一次输出未通过 Schema 校验,错误如下:\n"+"\n".join(lines)+"\n\n请严格按以下 JSON Schema 重新输出,只输出 JSON,不要解释:\n"+json.dumps(schema,ensure_ascii=False))defcall_structured(llm,system:str,user:str,model_cls:type[BaseModel],max_retry:int=3,temperature:float=0.0):schema=model_cls.model_json_schema()messages=[{"role":"system","content":system},{"role":"user","content":user}]last_err:Exception|None=Noneforattemptinrange(max_retry+1):resp=llm(messages,response_format={"type":"json_object"},temperature=temperature)raw=extract_json(resp.text)orresp.texttry:returnmodel_cls.model_validate_json(raw)except(ValidationError,ValueError)ase:last_err=e# 关键:把失败输出和具体错误一起回填messages.append({"role":"assistant","content":resp.text})messages.append({"role":"user","content":_repair_prompt(e,schema)})raiseStructuredOutputError(f"重试{max_retry}次仍失败:{last_err}")注意几个细节:temperature=0.0降低随机性;只回填前 5 条错误(全量错误会挤爆上下文,且让模型抓不住重点);把模型的原始错误输出作为assistant消息回填,保持对话结构合法。
extract_json返回None时用resp.text兜底,是为了让 Pydantic 抛出更具体的 JSON 解析错误,而不是直接丢失原始文本。model_validate_json会同时检查 JSON 语法和 Schema 约束,捕获ValidationError和ValueError。在重试时,我们并没有丢弃上一次的失败输出,而是把它作为assistant消息追加到对话历史中。这样做有两个好处:一是保持消息角色交替的合法性,二是让模型看到自己刚才错在哪,便于针对性修正。
3.3 生产级封装:异步、并发、监控与降级
上面的主循环已经能跑通,但直接放进生产环境还不够。我们需要考虑异步调用、并发控制、日志监控和降级策略。
3.3.1 异步版本
很多 LLM SDK 都提供异步接口。把call_structured改造成异步版本,可以在高并发场景下显著提升吞吐:
importasyncioasyncdefcall_structured_async(llm,system:str,user:str,model_cls:type[BaseModel],max_retry:int=3,temperature:float=0.0):schema=model_cls.model_json_schema()messages=[{"role":"system","content":system},{"role":"user","content":user}]last_err=Noneforattemptinrange(max_retry+1):resp=awaitllm(messages,response_format={"type":"json_object"},temperature=temperature)raw=extract_json(resp.text)orresp.texttry:returnmodel_cls.model_validate_json(raw)except(ValidationError,ValueError)ase:last_err=e messages.append({"role":"assistant","content":resp.text})messages.append({"role":"user","content":_repair_prompt(e,schema)})raiseStructuredOutputError(f"重试{max_retry}次仍失败:{last_err}")3.3.2 并发控制与重试风暴
如果一次性并发几百个请求,每个请求失败后又重试三次,很容易把 LLM API 的速率限制打满,甚至触发封禁。建议用asyncio.Semaphore控制并发数,并对重试增加指数退避:
importrandomasyncdefcall_with_backoff(llm,system,user,model_cls,sem:asyncio.Semaphore,max_retry=3):asyncwithsem:forattemptinrange(max_retry+1):try:returnawaitcall_structured_async(llm,system,user,model_cls,max_retry=0)exceptStructuredOutputError:ifattempt==max_retry:raisewait=min(2**attempt+random.random(),8)awaitasyncio.sleep(wait)3.3.3 日志与监控
生产环境必须记录每次调用的尝试次数、失败原因、耗时和最终结果。可以用结构化日志:
importlogging,time logger=logging.getLogger("structured_output")deflog_attempt(request_id,attempt,success,error_type=None,latency=None):logger.info({"request_id":request_id,"attempt":attempt,"success":success,"error_type":error_type,"latency_ms":latency,})关键指标包括:首次成功率、重试后成功率、平均重试次数、失败原因分布。如果首次成功率突然下降,可能是模型版本更新或 prompt 被改坏;如果重试后仍失败率升高,可能需要调整 Schema 或增加约束解码。
3.3.4 与不同 LLM API 的集成
- OpenAI:支持
response_format={"type": "json_schema", "json_schema": {...}}的 Structured Outputs,也支持json_object。优先用json_schema,能获得解码级约束。 - Anthropic Claude:没有原生 JSON Schema 约束,但可以通过 tool use 强制输出 JSON,再用 Pydantic 校验。
- 国产模型(通义千问、智谱、百川等):部分支持
response_format={"type": "json_object"},部分只支持 prompt 约束。对于不支持约束解码的模型,提取器 + 校验 + 重试就是主力。 - 本地模型:可以用
outlines、llguidance、XGrammar做受约束解码。例如outlines.generate.json(model, schema)可以直接生成符合 Schema 的 JSON。
3.3.5 缓存与幂等
如果相同的输入会被多次调用,可以在网关层加缓存。缓存键由system + user + model_cls.__name__的哈希组成。对于失败重试,缓存的是最终成功的结果,而不是中间失败。注意,如果模型版本或 Schema 变了,要主动失效缓存。
3.3.6 降级策略
当重试耗尽仍然失败时,不要直接抛异常给用户。可以按业务重要性降级:
- 返回默认值,并标记
need_human=True; - 切换到备用模型或更小的模型;
- 把原始文本和错误信息写入死信队列,由人工处理;
- 对于非关键字段,允许部分缺失,只要核心字段完整。
四、常见问题 FAQ
Q1:开了 JSON Mode 为什么还会格式错误?
JSON Mode 只保证输出是合法 JSON,不保证符合你的字段结构。模型可能返回{"result": "..."}而不是你期望的{"title": ..., "priority": ...}。另外,部分厂商的 JSON Mode 实现并不完善,仍可能输出 markdown 代码块或前后缀。因此,提取器和 Pydantic 校验不能省。
Q2:Pydantic 校验失败后,模型为什么还是不改?
可能原因有三:一是错误信息太长,模型抓不住重点;二是 Schema 太复杂,模型理解不了;三是模型本身能力不足。建议限制回填错误条数(如 5 条),把 Schema 精简到必要字段,并换用支持 Structured Outputs 的模型。
Q3:重试次数设多少合适?
通常 2~3 次。首次失败后,第一次重试成功率最高;第二次重试收益递减;超过三次后,失败往往源于模型能力或 Schema 问题,继续重试只会浪费 token 和时间。建议设置max_retry=3,并在第三次失败后降级。
Q4:如何让模型输出嵌套结构?
Pydantic 支持嵌套模型。例如class Order(BaseModel): items: list[Item]。生成的 JSON Schema 会包含嵌套定义。注意,嵌套越深,模型越容易出错,建议把嵌套层数控制在 3 层以内。
Q5:大模型返回的 JSON 里有中文怎么办?
Python 的json.loads和 Pydantic 都能正确处理 Unicode。如果遇到编码问题,确保resp.text是 UTF-8 字符串,不要先.encode()再.decode()。写入日志时使用ensure_ascii=False便于排查。
Q6:如何处理长文本截断导致的 JSON 不完整?
首先设置合理的max_tokens,预留足够空间。其次,如果finish_reason是length,直接判定失败并重试,不要尝试补全。也可以在 prompt 中要求模型“先输出简短摘要,再输出 JSON”,减少截断概率。
Q7:约束解码(Structured Outputs)支持所有模型吗?
不支持。OpenAI 的部分模型、开源 outlines 支持的模型可以。国产模型支持情况不一。对于不支持的模型,只能靠提取、校验和重试。选型时优先选支持 Schema 约束的接口。
Q8:使用 LangChain 的 PydanticOutputParser 够吗?
LangChain 的 PydanticOutputParser 提供了 prompt 模板和解析,但重试机制较弱。它通常只做一次解析,失败就抛错。生产环境建议自己实现回填重试循环,或者使用 LangChain 的with_retry并自定义错误处理。
Q9:如何做单元测试和集成测试?
单元测试:用固定的脏文本测试extract_json,用非法 JSON 测试 Pydantic 校验,用模拟 LLM 测试重试逻辑。集成测试:用真实 API 跑一批样本,统计首次成功率和重试后成功率。把失败样本加入回归集,防止 prompt 或 Schema 变更导致退化。
Q10:生产环境如何降级?
见 3.3.6。核心原则是:不要让结构化输出失败直接变成用户可见的错误。可以返回默认值、标记人工复核、写死信队列、切换备用模型。同时记录足够的日志,便于事后分析。
五、踩坑与优化建议
- 不要在 system prompt 里堆砌过多格式要求。指令太长会稀释重点,模型反而记不住。把 Schema 放在单独的 user 消息或 response_format 里更有效。
- 不要用
json.loads()直接解析。先extract_json,再model_validate_json。前者处理脏文本,后者处理类型和约束。 - 不要把错误全量回填。Pydantic 可能返回几十条错误,全塞给模型会挤爆上下文。限制前 5 条,按字段重要性排序。
- temperature 不要设太高。结构化输出场景建议 0.0~0.3。高温会增加格式漂移和枚举幻觉。
- 注意
max_tokens。如果输出较长,务必设置足够大的max_tokens,并检查finish_reason。截断的 JSON 无法修复。 - 注意 markdown 代码块。即使模型承诺只输出 JSON,也可能加 ```json。提取器必须处理。
- 注意 JSON 中的 NaN/Infinity。Python 的
json.loads默认允许NaN,但标准 JSON 不允许。可以用parse_constant拒绝,或在 Pydantic 中禁用。 - 注意 Optional 字段与默认值。如果字段可选,模型可能省略,也可能返回
null。Schema 要明确anyOf或default。 - 并发重试要加退避。否则失败请求会瞬间放大,触发 API 限流。
- 使用
additionalProperties: false。在 JSON Schema 中禁止额外字段,可以防止模型返回无关内容。 - 监控失败率并设置告警。首次成功率、重试后成功率、平均重试次数都是关键指标。
- 优先使用约束解码。如果模型支持 Structured Outputs,直接用它,能把大量失败消灭在解码阶段。
六、总结与最佳实践
回到最初的问题:大模型输出总是不符合格式,怎么办?答案是把它当作工程问题,而不是 prompt 问题。三层防线缺一不可:
- 语法层:用 JSON Mode 或约束解码,保证输出是合法 JSON,最好符合 Schema。
- 语义层:用 Pydantic 定义契约,做运行时校验,把业务约束写进类型。
- 失败层:用错误回填重试,把校验错误变成模型可理解的反馈,并设置重试上限和降级策略。
最后给出一份 checklist,供你在接入 LLM 时逐项核对:
- 是否定义了 Pydantic 模型,并用
model_json_schema()生成 Schema? - 是否优先使用支持 Structured Outputs 的接口?
- 是否实现了健壮的
extract_json,能处理代码块、前后缀、嵌套括号? - 是否在重试时回填了具体错误,并限制错误条数?
- 是否设置了
temperature=0.0或较低值? - 是否设置了合理的
max_tokens并检查截断? - 是否记录了尝试次数、失败原因和耗时?
- 是否有降级策略,避免重试耗尽后直接报错?
- 是否对并发重试做了退避和限流?
- 是否定期统计首次成功率和重试后成功率?
把这套网关封装成内部库,所有需要结构化输出的业务都复用它,你就能把“模型输出不是 JSON”从一个每天折磨人的玄学问题,变成一个可监控、可重试、可降级的工程组件。
更多硬核网安与AI工具包,请扫码获取完整源码!