「模型偶尔返回不是 JSON 的东西」是 AI 应用接后端最头疼的问题。这篇把可靠拿到结构化输出的三层防线讲透约束解码、Schema 校验、失败重试代码可直接用。文章目录一、为什么这是刚需二、第一层用 Schema 定义「标准答案」三、第二层三种拿到结构化输出的方式方式 A原生 Structured Output首选若模型支持方式 BFunction Calling / Tool Use兼容性最好方式 CPrompt 约束 稳健解析通用兜底四、第三层校验失败后的重试关键的一层五、进阶批量抽取与并发控制六、结构化输出的准确率提升清单七、踩坑清单八、上线前检查清单一、为什么这是刚需只要你的 AI 应用要和数据库、前端、其他服务对接就必然需要结构化输出场景需要的数据工单自动分类{category: 物流, priority: 2}简历解析{name: ..., skills: [...], years: 3}智能体工具调用{tool: search, params: {...}}数据抽取{company: ..., amount: 12000, date: 2026-09-29}而模型的默认输出是自然语言它会热情地给你加解释、加 Markdown 代码块、字段名偶尔改成中文——后端json.loads()直接崩。解决思路是三层防线尽量约束 → 严格校验 → 优雅重试。少一层都不够稳。二、第一层用 Schema 定义「标准答案」先定义数据长什么样这是所有后续防线的基准# schemas.pyfrompydanticimportBaseModel,Field,field_validatorfromtypingimportLiteralfromenumimportEnumclassCategory(str,Enum):LOGISTICS物流QUALITY质量AFTERSALE售后OTHER其他classTicketAnalysis(BaseModel):category:CategoryField(...,description工单分类)priority:intField(...,ge1,le5,description优先级 1最低 5最高)summary:strField(...,max_length100,description一句话摘要不超过100字)need_human:boolField(...,description是否需要人工介入)field_validator(summary)classmethoddefno_emoji(cls,v:str)-str:# 自定义校验摘要里不许有表情符号下游系统存不下ifany(ord(c)0x1F000forcinv):raiseValueError(摘要不允许包含表情符号)returnv两个要点每个字段都写description——它会进 JSON Schema 给模型看描述写得准模型填得对用Enum和Literal而不是裸str——把可选值钉死是提高准确率最有效的一招。三、第二层三种拿到结构化输出的方式方式 A原生 Structured Output首选若模型支持fromopenaiimportOpenAI clientOpenAI()respclient.beta.chat.completions.parse(modelgpt-4.1-mini,messages[{role:user,content:f分析这条工单{ticket_text}}],response_formatTicketAnalysis,# 直接传 Pydantic 模型)result:TicketAnalysisresp.choices[0].message.parsed# 已经是对象不是字符串print(result.category,result.priority)这是最省事的方式模型端在解码时就保证输出符合 Schema理论上不会产出非法 JSON。有就用它没有就走下面两种。方式 BFunction Calling / Tool Use兼容性最好importjsonfromopenaiimportOpenAI clientOpenAI()tools[{type:function,function:{name:submit_analysis,description:提交工单分析结果,parameters:TicketAnalysis.model_json_schema(),# Pydantic 直接生成 Schema},}]respclient.chat.completions.create(modelqwen3:8b,# 本地模型也支持Ollama/vLLM 均兼容messages[{role:user,content:f分析这条工单{ticket_text}}],toolstools,tool_choice{type:function,function:{name:submit_analysis}},# 强制调用)argsjson.loads(resp.choices[0].message.tool_calls[0].function.arguments)resultTicketAnalysis.model_validate(args)tool_choice强制指定函数能显著减少模型「先聊两句」的概率。方式 CPrompt 约束 稳健解析通用兜底老模型、本地小模型可能两种都不支持那就靠 Prompt 容错解析SYSTEM你是工单分析器。只输出一个 JSON 对象不要任何解释、不要 Markdown 代码块。 字段category(物流/质量/售后/其他), priority(1-5整数), summary(≤100字), need_human(布尔) 示例输出 {category: 物流, priority: 3, summary: 用户催单快递停滞三天, need_human: false}defparse_loose(text:str)-TicketAnalysis:容错解析处理模型最常见的三种「不听话」ttext.strip()# 1) 剥掉 Markdown 代码块 json ... ift.startswith():tt.split()[1]ift.lower().startswith(json):tt[4:]tt.strip()# 2) 截取第一个 { 到最后一个 }丢掉前后的客套话if{intand}int:tt[t.index({):t.rindex(})1]# 3) 常见修复True/False 首字母大写、中文引号tt.replace(True,true).replace(False,false)tt.replace(“,).replace(”,)returnTicketAnalysis.model_validate_json(t)这个parse_loose看着土但在真实项目里能救回大量请求推荐所有走 Prompt 的项目都配上。四、第三层校验失败后的重试关键的一层真实场景一定会有失败。正确的重试姿势是把错误信息喂回给模型让它自己修而不是原样重发# robust_extract.pyimportjsonfromopenaiimportOpenAIfrompydanticimportValidationErrorfromschemasimportTicketAnalysis,Category# 复用第二节定义clientOpenAI()defextract(text:str,model:strgpt-4.1-mini,max_retries:int2)-TicketAnalysis:messages[{role:system,content:SYSTEM},{role:user,content:f分析这条工单{text}},]last_errorNoneforattemptinrange(max_retries1):respclient.chat.completions.create(modelmodel,messagesmessages,temperature0,# 抽取任务一律 temperature0)rawresp.choices[0].message.contenttry:returnparse_loose(raw)except(ValidationError,json.JSONDecodeError,ValueError)ase:last_errore# 把原始输出和具体错误一起回灌 —— 这一步是重试成功率的来源messages[{role:assistant,content:raw},{role:user,content:f上面的输出不符合要求错误{e}\nf请重新只输出合法 JSON字段要求{SYSTEM.splitlines()[-1]}},]# 全部失败 → 走业务兜底而不是抛异常把上游打挂returnTicketAnalysis(categoryCategory.OTHER,priority1,summary自动解析失败请人工处理,need_humanTrue)四个关键设计temperature0抽取任务不需要创造力温度只会带来不确定性错误回灌把校验错误原文告诉模型重试成功率通常能到 80% 以上只重发不告诉错在哪成功率提升有限有上限max_retries2就够了无限重试只会烧钱有兜底返回值宁可返回一个「需人工处理」的默认对象也不要让异常穿透到业务层。五、进阶批量抽取与并发控制要处理一万条工单一条条同步调太慢。用异步 信号量# batch_extract.pyimportasynciofromopenaiimportAsyncOpenAI aclientAsyncOpenAI()asyncdefextract_one(text:str,sem:asyncio.Semaphore)-dict:asyncwithsem:# 控制并发避免被限流try:rawaitasyncio.to_thread(extract,text)# 复用上面的同步实现return{input:text,output:r.model_dump()}exceptExceptionase:return{input:text,error:str(e)}asyncdefbatch(texts:list[str],concurrency:int8):semasyncio.Semaphore(concurrency)returnawaitasyncio.gather(*[extract_one(t,sem)fortintexts])并发数按你的 API 限额定保守起步5-10、观察 429 报错再调。比起盲目开 50 并发被限流封禁稳一点划算得多。六、结构化输出的准确率提升清单按收益从高到低排手段提升幅度成本用 Enum/Literal 钉死可选值高零每个字段写清 description高零给 1-2 个完整示例Few-shot高少量 tokentemperature0中高零原生 Structured Output / 强制 tool_choice中高需模型支持错误回灌重试中对剩余失败样本极高少量 token换更大模型中贵先做前四条——零成本的那几条做到位多数场景准确率能到 95%根本不需要换模型。七、踩坑清单现象原因解决返回带 json 代码块模型习惯了 Markdownparse_loose剥离或改用 Structured Output字段名变成中文Prompt 里字段没写死显式给出示例输出的字段名数字返回成字符串3未约束类型Pydantic 声明int会自动转换但要在 Schema 里写清integer枚举值给了同义词没列全可选值用 Enum description 列全重试仍然失败只重发没给错误信息错误回灌第四节本地小模型完全不支持 JSON能力限制换 14B 以上或降级用规则抽取八、上线前检查清单所有抽取类请求temperature0Schema 用 Enum/Literal 约束枚举字段每个字段有 description有容错解析函数处理 Markdown 包裹 / 前后废话 / 大小写重试带错误回灌上限 2 次全部失败有业务兜底对象不向上抛异常批量任务有并发控制盯 429抽取失败率有监控正常应 5%持续偏高说明 Prompt 或 Schema 要改。结构化输出是 AI 应用从「演示」走向「生产」最基础的一道工程关。这三层防线搭好你的 AI 服务才算真的能接进业务系统。