如何让大语言模型稳定输出 JSON 的三层防御体系

核心思想从“灵光一闪”到“确定防线”不能让模型靠运气输出要用软件工程防线包裹非确定性的生成模型最终实现条件反射式的正确输出。对于极致场景可通过SFT 监督微调让模型形成“肌肉记忆”。三层防线详解第一道防线提示词控制软约束Schema 注入 Few-Shot提供 JSON Schema 或 TypeScript 接口定义并给 2-3 个范例把字段类型约束写死。利用近因效应指令末尾强调“禁绝废话仅输出 JSON”并用定界符json ...框定区域。第二道防线生成控制层物理硬手段闭源 API使用Structured Outputs功能预编译 Schema 进解码引擎遵循率接近 100%。开源模型采用Logit Masking概率掩码通过有限状态机强制将不符合语法的 token 概率设为负无穷杜绝格式错误。第三道防线工程校验与自修复闭环逻辑强校验框架用Pydantic等工具检查字段完整性和类型正确性杜绝脏数据。流式解析优化对超长 JSON 边生成边检查发现错误立即中断节省算力。闭环自修复将校验报错信息如“字段缺失”反馈给模型指导其利用 LLM 能力自我修正。第一部分第三道防线 —— Pydantic 强校验与自修复这里展示一个完整的闭环逻辑定义结构 - 尝试解析 - 失败则把报错喂给模型重试。import json import openai from pydantic import BaseModel, Field, ValidationError from typing import List, Optional # 1. 定义期望的数据结构 (Schema) class UserProfile(BaseModel): name: str Field(description用户姓名) age: int Field(description年龄) skills: List[str] Field(description技能列表) email: Optional[str] Field(description邮箱可选) # 2. 模拟大模型返回的“不干净”数据 (JSON字符串) bad_response_from_llm { name: 张三, age: 二十八, skills: [Python, Java], extra_field: 这是多余字段会被忽略 } def validate_and_repair(llm_output: str, max_retries: int 2): 校验失败时将报错反馈给模型进行自修复 for attempt in range(max_retries): try: # 解析 JSON 字符串 data json.loads(llm_output) # 第三道防线核心Pydantic 校验 # 这里会自动检查 age 是否为 int缺少字段会报错多余字段默认忽略 validated_data UserProfile(**data) print(f校验通过数据: {validated_data.model_dump()}) return validated_data except (json.JSONDecodeError, ValidationError) as e: print(f第 {attempt 1} 次校验失败: {e}) if attempt max_retries - 1: raise e # 闭环自修复机制 # 把具体的报错信息 (如 age: Input should be a valid integer) # 反馈给模型让模型自己重写 JSON repair_prompt f 你之前输出的 JSON 格式有误报错信息如下 {str(e)} 请根据以下 Schema 定义修正错误仅输出纯 JSON不要解释 {UserProfile.model_json_schema()} 原始错误输出 {llm_output} # 这里调用 LLM (以 OpenAI 为例) # 注意实际生产中这里需要调用你的模型 API print(正在请求模型自我修复...) # response openai.ChatCompletion.create(...) # llm_output response.choices[0].message.content # 模拟修复后的输出 (实际运行时会替换为模型返回) llm_output {name: 张三, age: 28, skills: [Python, Java], email: null} return None # 执行测试 validate_and_repair(bad_response_from_llm)第二部分第二道防线 —— OpenAI Structured Outputs 写法这是物理硬手段在 API 请求阶段就直接约束模型生成符合 Schema 的 Token。from openai import OpenAI from pydantic import BaseModel from typing import List client OpenAI(api_keyyour-api-key) # 1. 定义结构 (使用 Pydantic 定义即可OpenAI SDK 直接兼容) class Step(BaseModel): explanation: str output: str class MathResponse(BaseModel): steps: List[Step] final_answer: str # 2. 调用 API使用 parse 方法 (Structured Outputs 模式) completion client.beta.chat.completions.parse( modelgpt-4o-2024-08-06, # 注意需使用支持 Structured Outputs 的模型版本 messages[ {role: system, content: 你是一个数学老师回答问题要分步骤。}, {role: user, content: 解方程 2x 5 15}, ], response_formatMathResponse, # 直接传入 Pydantic 类 ) # 3. 获取结果 —— 已经是实例化好的 Pydantic 对象100% 符合结构 math_result completion.msg.parsed # 4. 直接使用属性无需担心格式错误 print(f最终答案: {math_result.final_answer}) for i, step in enumerate(math_result.steps): print(f步骤 {i1}: {step.explanation} - {step.output}) # 如果是开源模型这里本应是 Logit Masking 的代码 # 开源侧通常使用 outlines、guidance 或 llama.cpp grammars 库 # 例如 outlines 示例 (伪代码逻辑): # import outlines # model outlines.models.transformers(microsoft/Phi-3-mini-4k-instruct) # --- 构建 Logit Masking 生成器 (第二道防线物理硬手段) --- # 这一行代码背后Outlines 将 Pydantic Schema 编译成了有限状态机 (FSM) # 并在推理时自动挂载 LogitsProcessor 进行概率掩码。 # generator outlines.generate.json(model, UserProfile.model_json_schema()) # result generator(给我一个张三的用户档案)总结两者的区别维度Pydantic 校验 (第三道防线)OpenAI Structured Outputs (第二道防线)执行时机模型生成之后模型生成过程之中类比质检员带卡槽的模具适用场景所有模型、老版本 API、极端自定义校验闭源商业 API (OpenAI, 火山等)优点灵活能捕获业务逻辑错误能自修复零格式错误速度快Token 省一个综合运用三道防线的完整生产级示例。场景设定为AI 辅助医疗问诊需要模型输出包含诊断和处方的严格 JSON。将代码分为四个逻辑块第一道防线Prompt构建、第二道防线API约束尝试、第三道防线Pydantic校验与自修复。import json import openai from pydantic import BaseModel, Field, ValidationError, field_validator from typing import List, Optional, Literal from openai import OpenAI # 0. 初始化与Schema定义 client OpenAI(api_keyyour-api-key) # 请替换为真实Key # 定义处方的药物结构 class PrescriptionDrug(BaseModel): name: str Field(..., description药物通用名) dosage: str Field(..., description单次剂量如 10mg) frequency: str Field(..., description用药频率如 每日一次) duration: str Field(..., description疗程如 7天) # 定义完整的诊断报告结构Pydantic Schema用于第一道防线的注入和第三道防线的校验 class DiagnosisReport(BaseModel): patient_complaint: str Field(..., description主诉总结患者问题) # 使用 Literal 限制枚举值防止模型瞎编严重程度 severity: Literal[轻度, 中度, 重度, 危急] Field(..., description严重程度分级) diagnosis: str Field(..., description初步诊断名称) reasoning: str Field(..., description诊断依据逻辑推理过程) prescriptions: List[PrescriptionDrug] Field(..., min_length1, description处方列表至少包含一种药物或处理方式) notes: Optional[str] Field(None, description医嘱备注如饮食禁忌) # 第三道防线补充自定义逻辑校验Pydantic 的 field_validator field_validator(reasoning) def reasoning_must_be_meaningful(cls, v): if len(v) 10: raise ValueError(诊断依据过于简短需详细说明) return v # 1. 第一道防线Prompt 工程 def build_defense_prompt(user_input: str) - str: 应用高手策略 1. Schema注入 Few-Shot (给出一个范例) 2. 近因效应 (末尾强调格式) # 将 Pydantic Schema 转换为 JSON 结构描述字符串注入给模型看 schema_desc DiagnosisReport.model_json_schema() system_prompt f 你是一名专业的全科医生AI助手。你必须严格按照以下 JSON Schema 定义输出诊断结果。 Schema 定义如下 {schema_desc} 【Few-Shot 范例】 用户输入我头疼流鼻涕有点发烧38度 正确输出 {{ patient_complaint: 头痛、流涕伴发热1天, severity: 中度, diagnosis: 急性上呼吸道感染感冒, reasoning: 患者有明确的鼻部卡他症状流涕及发热体征符合病毒性上呼吸道感染典型表现。, prescriptions: [ {{ name: 对乙酰氨基酚片, dosage: 500mg, frequency: 必要时服用, duration: 3天 }}, {{ name: 生理性海水鼻腔喷雾, dosage: 每侧2喷, frequency: 每日3次, duration: 5天 }} ], notes: 多喝温水注意休息若高热不退或出现呼吸困难请及时就医。 }} user_prompt f 患者描述{user_input} 请给出诊断和处方。 # 第一道防线核心近因效应 # 将最严格的格式约束指令放在 Prompt 的最末尾 final_instruction 【最高优先级指令】 1. 仅输出符合上述 JSON Schema 的纯 JSON 字符串。 2. 严禁在 JSON 之外添加任何解释、问候语、Markdown标记如 json或换行字符。 3. 确保所有字段齐全prescriptions 必须为数组。 # 拼接最终消息 messages [ {role: system, content: system_prompt}, {role: user, content: user_prompt final_instruction} ] return messages # 2. 第二道防线生成控制层 (可选硬手段) def call_llm_with_structured_output(messages): 尝试使用 OpenAI Structured Outputs 作为第二道防线物理硬手段。 如果模型或API不支持则降级为普通JSON模式调用。 try: # 使用 beta.chat.completions.parse 进行强约束生成 completion client.beta.chat.completions.parse( modelgpt-4o-2024-08-06, # 确保是支持此功能的模型 messagesmessages, response_formatDiagnosisReport, temperature0.1 # 降低随机性提高格式遵循度 ) # 如果成功这里返回的已经是校验通过的 Pydantic 对象 return completion.choices[0].message.parsed, None except Exception as e: # 降级逻辑如果不支持 Structured Outputs 或网络出错走普通 Chat 接口 print(f第二道防线未启用或失败 (降级至普通生成): {e}) response client.chat.completions.create( modelgpt-3.5-turbo, # 普通模型 messagesmessages, temperature0.1 ) raw_content response.choices[0].message.content return None, raw_content # 3. 第三道防线Pydantic 校验 闭环自修复 def robust_diagnosis_parser(raw_json_str: str, original_messages: list) - DiagnosisReport: 尝试解析 JSON失败则进入重试循环自修复。 MAX_RETRY 2 current_output raw_json_str for attempt in range(MAX_RETRY 1): try: # 1. 基础清洗移除可能的 Markdown 代码块标记 (第一道防线失效时的补救) clean_str current_output.strip() if clean_str.startswith(json): clean_str clean_str[7:] if clean_str.startswith(): clean_str clean_str[3:] if clean_str.endswith(): clean_str clean_str[:-3] # 2. 解析 JSON data_dict json.loads(clean_str) # 3. Pydantic 强校验 (字段缺失、类型错误、自定义校验器都会在此报错) validated_report DiagnosisReport(**data_dict) print(f第三道防线校验通过 (尝试次数: {attempt 1})) return validated_report except (json.JSONDecodeError, ValidationError) as e: print(f解析/校验失败 (Attempt {attempt 1}): {e}) if attempt MAX_RETRY: raise ValueError(f模型经过 {MAX_RETRY} 次修复仍未通过校验最后错误: {e}) # 闭环自修复逻辑 # 构建修复用的消息直接追加报错信息和错误输出 repair_messages original_messages.copy() # 模拟 Assistant 的错误回答 repair_messages.append({role: assistant, content: current_output}) # 用户反馈错误详情 (这是最关键的反馈信号) user_feedback f 你的上一轮回答 JSON 格式或内容校验失败详细错误如下 {str(e)} 请根据报错信息修正 JSON。确保 1. severity 只能是 轻度、中度、重度、危急 之一。 2. reasoning 长度需大于10个字符。 3. 字段必须完整。 请直接输出修正后的纯 JSON不要加任何多余字符。 repair_messages.append({role: user, content: user_feedback}) print(进入自修复模式重新请求模型...) # 重新调用模型 (这里为了演示直接走普通接口实际可复用 call_llm_with_structured_output) response client.chat.completions.create( modelgpt-3.5-turbo, messagesrepair_messages, temperature0.0 ) current_output response.choices[0].message.content # 4. 主流程整合 (三道防线协同) def get_medical_diagnosis(user_query: str): print(f用户输入: {user_query}) # Step 1: 构建第一道防线 Prompt messages build_defense_prompt(user_query) # Step 2: 尝试第二道防线 (物理硬约束) parsed_obj, raw_text call_llm_with_structured_output(messages) # Step 3: 如果第二道防线直接成功了 (返回了 Pydantic 对象) if parsed_obj is not None: print(第二道防线直接命中无需第三道防线介入。) return parsed_obj # Step 4: 否则进入第三道防线 (软件校验 自修复) print(启动第三道防线工程校验与自修复...) final_report robust_diagnosis_parser(raw_text, messages) return final_report # 运行示例 if __name__ __main__: # 模拟一个可能让模型出错的模糊输入 test_input 嗓子疼浑身没劲感觉发冷量了下38度5 try: result get_medical_diagnosis(test_input) print(\n 最终处方单 (JSON) ) # 直接打印格式化的 JSON完全符合应用层调用要求 print(result.model_dump_json(indent2, ensure_asciiFalse)) except Exception as e: print(f系统处理异常: {e})三道防线在这个例子中的具体体现防线层级代码位置具体作用对应策略第一道build_defense_prompt1. 注入model_json_schema()2. 提供 Few-Shot 范例3.末尾强调“仅输出 JSON”Schema注入 近因效应第二道call_llm_with_structured_output使用client.beta.chat.completions.parseStructured Outputs(物理硬手段)第三道robust_diagnosis_parser1.Pydantic校验字段类型、枚举值2.field_validator校验逻辑3. 失败时将报错反馈给模型强校验 闭环自修复运行预期输出当模型第一次尝试输出severity: 高烧不在枚举范围或缺失字段时系统会打印解析/校验失败 (Attempt 1): 1 validation error for DiagnosisReport severity ... 进入自修复模式重新请求模型... 第三道防线校验通过 (尝试次数: 2)最终输出一定是符合DiagnosisReport定义的干净 JSON可以直接被下游的业务代码如电子病历系统、药房系统使用。