大模型结构化输出与 JSON Schema 约束生成从自由文本到可靠数据一、大模型输出的自由散漫为什么你的 JSON 总是解析失败大模型天生是自由文本生成器——它擅长写文章、讲故事但不擅长输出严格格式的结构化数据。当你要求大模型输出 JSON 时它可能给你加上 Markdown 代码块标记、在键名前后加空格、漏掉闭合括号、甚至在中途跑题生成一段解释文字。在 Agent 系统中下游工具依赖 JSON 格式的输入一次解析失败就可能导致整个工作流中断。结构化输出技术通过约束解码Constrained Decoding和 JSON Schema 约束确保大模型的每一步生成都符合预定义的格式规范将自由文本转化为可靠数据。二、结构化输出架构flowchart TD A[用户请求] -- B[JSON Schema 定义] B -- C[约束解码引擎] C -- C1[Token 白名单过滤] C -- C2[状态机驱动生成] C -- C3[格式修复兜底] C1 -- D[结构化 JSON 输出] C2 -- D C3 -- D D -- E[输出校验] E -- E1[Schema 校验] E -- E2[语义校验] E1 -- F[可靠数据] E2 -- F2.1 JSON Schema 定义与约束# schema_definitions.py — 常用 JSON Schema 定义 # 设计意图为大模型结构化输出提供标准化的 Schema 定义 # API 响应 Schema API_RESPONSE_SCHEMA { type: object, properties: { status: {type: string, enum: [success, error]}, data: {type: object}, error: { type: [string, null], description: 错误信息成功时为 null, }, }, required: [status, data], } # 工具调用 Schema TOOL_CALL_SCHEMA { type: object, properties: { tool: {type: string, description: 工具名称}, arguments: { type: object, description: 工具参数, additionalProperties: True, }, thought: { type: string, description: 调用此工具的推理过程, }, }, required: [tool, arguments], } # 数据提取 Schema EXTRACT_ENTITIES_SCHEMA { type: object, properties: { entities: { type: array, items: { type: object, properties: { name: {type: string}, type: { type: string, enum: [person, organization, location, date, product], }, confidence: { type: number, minimum: 0, maximum: 1, }, }, required: [name, type, confidence], }, } }, required: [entities], }2.2 约束解码引擎# constrained_decoder.py — 约束解码引擎 # 设计意图在生成过程中约束每一步的 token 选择确保输出符合 JSON 格式 import json import re from typing import Generator class JSONConstrainedDecoder: 基于状态机的 JSON 约束解码器 核心思想维护 JSON 生成的当前状态对象键、数组元素、字符串值等 在每一步只允许生成符合当前状态的 token def __init__(self, schema: dict): self.schema schema self.state_stack [start] # 状态栈 self.key_stack [] # 当前键路径 def get_allowed_tokens(self, generated_text: str) - list[str] | None: 根据已生成的文本返回下一步允许的 token 返回 None 表示不约束自由生成 返回空列表表示无合法 token格式错误 # 尝试解析已生成的文本判断当前状态 text generated_text.strip() # 移除 Markdown 代码块标记 text re.sub(r^json\s*, , text) text re.sub(r\s*$, , text) # 判断是否在 JSON 字符串内部 if self._inside_json_string(text): return None # 字符串内部自由生成 # 判断当前 JSON 结构状态 if not text: return [{] # 必须以 { 开始 # 检查未闭合的括号 open_braces text.count({) - text.count(}) open_brackets text.count([) - text.count(]) if open_braces 0 or open_brackets 0: return None # JSON 结构未完成允许继续生成 return [] # JSON 已完成不允许继续生成 def _inside_json_string(self, text: str) - bool: 判断当前是否在 JSON 字符串内部 # 简化实现统计未闭合的引号数 in_string False escaped False for char in text: if escaped: escaped False continue if char \\: escaped True continue if char : in_string not in_string return in_string def post_process(self, text: str) - str: 后处理清理和修复 JSON 文本 # 移除 Markdown 代码块 text re.sub(r^json\s*, , text.strip()) text re.sub(r\s*$, , text) # 移除尾部非 JSON 字符 result depth 0 for char in text: if char {: depth 1 elif char }: depth - 1 result char if depth 0 and char }: break return result2.3 格式修复兜底# json_repair.py — JSON 格式修复 # 设计意图当约束解码无法完全保证格式时用修复逻辑兜底 import json import re class JSONRepair: JSON 格式修复器 def repair(self, text: str) - tuple[dict | None, str]: 尝试修复并解析 JSON 文本 返回: (解析结果, 修复说明) repairs [] # Step 1: 清理 text text.strip() text re.sub(r^json\s*, , text) text re.sub(r\s*$, , text) text text.strip() # Step 2: 直接解析 try: return json.loads(text), 无需修复 except json.JSONDecodeError: pass # Step 3: 修复常见问题 # 3.1 移除尾部逗号 text re.sub(r,\s*([}\]]), r\1, text) repairs.append(移除尾部逗号) # 3.2 补全缺失的闭合括号 open_braces text.count({) - text.count(}) open_brackets text.count([) - text.count(]) text ] * open_brackets } * open_braces if open_braces 0 or open_brackets 0: repairs.append(f补全 {open_braces} 个 }} 和 {open_brackets} 个 ]) # 3.3 修复单引号为双引号 text text.replace(, ) repairs.append(单引号替换为双引号) # Step 4: 再次解析 try: return json.loads(text), 修复: , .join(repairs) except json.JSONDecodeError as e: return None, f修复失败: {str(e)} def extract_json_from_text(self, text: str) - dict | None: 从混合文本中提取 JSON # 尝试匹配 JSON 对象 pattern r\{[^{}]*(?:\{[^{}]*\}[^{}]*)*\} for match in re.finditer(pattern, text, re.DOTALL): result, _ self.repair(match.group()) if result is not None: return result return None2.4 结构化输出 Pipeline# structured_output_pipeline.py — 结构化输出完整管线 # 设计意图整合约束解码、格式修复和 Schema 校验的完整管线 import json import jsonschema class StructuredOutputPipeline: def __init__(self, schema: dict, llm_client): self.schema schema self.llm_client llm_client self.decoder JSONConstrainedDecoder(schema) self.repair JSONRepair() async def generate(self, prompt: str, max_retries: int 3) - dict: 生成结构化输出 system_prompt self._build_system_prompt() for attempt in range(max_retries): # 生成 raw await self.llm_client.chat( f{system_prompt}\n\n{prompt}, temperature0.1, ) # 后处理 cleaned self.decoder.post_process(raw) # 修复 result, repair_msg self.repair.repair(cleaned) if result is None: # 修复失败重试并附加错误信息 prompt f\n\n[上次输出格式错误请确保输出合法 JSON] continue # Schema 校验 try: jsonschema.validate(result, self.schema) return result except jsonschema.ValidationError as e: prompt f\n\n[Schema 校验失败: {e.message}请修正] # 所有重试失败返回空结构 return self._empty_structure() def _build_system_prompt(self) - str: 构建系统提示词 schema_str json.dumps(self.schema, indent2, ensure_asciiFalse) return ( f你必须输出符合以下 JSON Schema 的 JSON 对象 f不要输出任何其他内容不要 Markdown 代码块标记不要解释文字\n fjson\n{schema_str}\n ) def _empty_structure(self) - dict: 生成符合 Schema 的空结构 if self.schema.get(type) object: result {} for key, prop in self.schema.get(properties, {}).items(): if key in self.schema.get(required, []): prop_type prop.get(type) if prop_type string: result[key] elif prop_type array: result[key] [] elif prop_type object: result[key] {} elif prop_type number: result[key] 0 elif prop_type boolean: result[key] False return result return {}四、边界分析与架构权衡约束解码的性能开销基于状态机的约束解码需要在每一步生成时计算允许的 token 集合增加约 10%-20% 的延迟。对于实时交互场景可以考虑使用 vLLM/Outlines 等框架的 GPU 加速约束解码。Schema 复杂度的限制嵌套层级过深或条件逻辑过复杂的 Schema如 oneOf、anyOf、if-then-else可能导致约束解码器状态爆炸。建议将复杂 Schema 拆分为多个简单 Schema分步生成。格式修复的可靠性修复逻辑基于启发式规则对于严重损坏的 JSON 可能修复出合法但语义错误的结果。建议在修复后增加语义校验如检查枚举值、数值范围。大模型的结构化能力差异不同模型对结构化输出的支持程度不同。GPT-4 和 Claude 3.5 的 JSON 生成准确率可达 95% 以上而开源 7B 模型可能只有 70%-80%。对于准确率要求高的场景建议使用支持原生结构化输出的模型 API如 OpenAI 的 Structured Outputs。五、总结大模型结构化输出通过约束解码、格式修复和 Schema 校验三层保障将自由文本转化为可靠数据。落地要点JSON Schema 定义输出格式规范约束解码器在生成过程中约束 token 选择格式修复兜底处理常见格式错误Schema 校验确保语义正确。关键权衡约束解码保证格式但增加延迟格式修复提高容错但可能引入语义错误简单 Schema 可靠但表达能力有限。
大模型结构化输出与 JSON Schema 约束生成：从“自由文本“到“可靠数据“
大模型结构化输出与 JSON Schema 约束生成从自由文本到可靠数据一、大模型输出的自由散漫为什么你的 JSON 总是解析失败大模型天生是自由文本生成器——它擅长写文章、讲故事但不擅长输出严格格式的结构化数据。当你要求大模型输出 JSON 时它可能给你加上 Markdown 代码块标记、在键名前后加空格、漏掉闭合括号、甚至在中途跑题生成一段解释文字。在 Agent 系统中下游工具依赖 JSON 格式的输入一次解析失败就可能导致整个工作流中断。结构化输出技术通过约束解码Constrained Decoding和 JSON Schema 约束确保大模型的每一步生成都符合预定义的格式规范将自由文本转化为可靠数据。二、结构化输出架构flowchart TD A[用户请求] -- B[JSON Schema 定义] B -- C[约束解码引擎] C -- C1[Token 白名单过滤] C -- C2[状态机驱动生成] C -- C3[格式修复兜底] C1 -- D[结构化 JSON 输出] C2 -- D C3 -- D D -- E[输出校验] E -- E1[Schema 校验] E -- E2[语义校验] E1 -- F[可靠数据] E2 -- F2.1 JSON Schema 定义与约束# schema_definitions.py — 常用 JSON Schema 定义 # 设计意图为大模型结构化输出提供标准化的 Schema 定义 # API 响应 Schema API_RESPONSE_SCHEMA { type: object, properties: { status: {type: string, enum: [success, error]}, data: {type: object}, error: { type: [string, null], description: 错误信息成功时为 null, }, }, required: [status, data], } # 工具调用 Schema TOOL_CALL_SCHEMA { type: object, properties: { tool: {type: string, description: 工具名称}, arguments: { type: object, description: 工具参数, additionalProperties: True, }, thought: { type: string, description: 调用此工具的推理过程, }, }, required: [tool, arguments], } # 数据提取 Schema EXTRACT_ENTITIES_SCHEMA { type: object, properties: { entities: { type: array, items: { type: object, properties: { name: {type: string}, type: { type: string, enum: [person, organization, location, date, product], }, confidence: { type: number, minimum: 0, maximum: 1, }, }, required: [name, type, confidence], }, } }, required: [entities], }2.2 约束解码引擎# constrained_decoder.py — 约束解码引擎 # 设计意图在生成过程中约束每一步的 token 选择确保输出符合 JSON 格式 import json import re from typing import Generator class JSONConstrainedDecoder: 基于状态机的 JSON 约束解码器 核心思想维护 JSON 生成的当前状态对象键、数组元素、字符串值等 在每一步只允许生成符合当前状态的 token def __init__(self, schema: dict): self.schema schema self.state_stack [start] # 状态栈 self.key_stack [] # 当前键路径 def get_allowed_tokens(self, generated_text: str) - list[str] | None: 根据已生成的文本返回下一步允许的 token 返回 None 表示不约束自由生成 返回空列表表示无合法 token格式错误 # 尝试解析已生成的文本判断当前状态 text generated_text.strip() # 移除 Markdown 代码块标记 text re.sub(r^json\s*, , text) text re.sub(r\s*$, , text) # 判断是否在 JSON 字符串内部 if self._inside_json_string(text): return None # 字符串内部自由生成 # 判断当前 JSON 结构状态 if not text: return [{] # 必须以 { 开始 # 检查未闭合的括号 open_braces text.count({) - text.count(}) open_brackets text.count([) - text.count(]) if open_braces 0 or open_brackets 0: return None # JSON 结构未完成允许继续生成 return [] # JSON 已完成不允许继续生成 def _inside_json_string(self, text: str) - bool: 判断当前是否在 JSON 字符串内部 # 简化实现统计未闭合的引号数 in_string False escaped False for char in text: if escaped: escaped False continue if char \\: escaped True continue if char : in_string not in_string return in_string def post_process(self, text: str) - str: 后处理清理和修复 JSON 文本 # 移除 Markdown 代码块 text re.sub(r^json\s*, , text.strip()) text re.sub(r\s*$, , text) # 移除尾部非 JSON 字符 result depth 0 for char in text: if char {: depth 1 elif char }: depth - 1 result char if depth 0 and char }: break return result2.3 格式修复兜底# json_repair.py — JSON 格式修复 # 设计意图当约束解码无法完全保证格式时用修复逻辑兜底 import json import re class JSONRepair: JSON 格式修复器 def repair(self, text: str) - tuple[dict | None, str]: 尝试修复并解析 JSON 文本 返回: (解析结果, 修复说明) repairs [] # Step 1: 清理 text text.strip() text re.sub(r^json\s*, , text) text re.sub(r\s*$, , text) text text.strip() # Step 2: 直接解析 try: return json.loads(text), 无需修复 except json.JSONDecodeError: pass # Step 3: 修复常见问题 # 3.1 移除尾部逗号 text re.sub(r,\s*([}\]]), r\1, text) repairs.append(移除尾部逗号) # 3.2 补全缺失的闭合括号 open_braces text.count({) - text.count(}) open_brackets text.count([) - text.count(]) text ] * open_brackets } * open_braces if open_braces 0 or open_brackets 0: repairs.append(f补全 {open_braces} 个 }} 和 {open_brackets} 个 ]) # 3.3 修复单引号为双引号 text text.replace(, ) repairs.append(单引号替换为双引号) # Step 4: 再次解析 try: return json.loads(text), 修复: , .join(repairs) except json.JSONDecodeError as e: return None, f修复失败: {str(e)} def extract_json_from_text(self, text: str) - dict | None: 从混合文本中提取 JSON # 尝试匹配 JSON 对象 pattern r\{[^{}]*(?:\{[^{}]*\}[^{}]*)*\} for match in re.finditer(pattern, text, re.DOTALL): result, _ self.repair(match.group()) if result is not None: return result return None2.4 结构化输出 Pipeline# structured_output_pipeline.py — 结构化输出完整管线 # 设计意图整合约束解码、格式修复和 Schema 校验的完整管线 import json import jsonschema class StructuredOutputPipeline: def __init__(self, schema: dict, llm_client): self.schema schema self.llm_client llm_client self.decoder JSONConstrainedDecoder(schema) self.repair JSONRepair() async def generate(self, prompt: str, max_retries: int 3) - dict: 生成结构化输出 system_prompt self._build_system_prompt() for attempt in range(max_retries): # 生成 raw await self.llm_client.chat( f{system_prompt}\n\n{prompt}, temperature0.1, ) # 后处理 cleaned self.decoder.post_process(raw) # 修复 result, repair_msg self.repair.repair(cleaned) if result is None: # 修复失败重试并附加错误信息 prompt f\n\n[上次输出格式错误请确保输出合法 JSON] continue # Schema 校验 try: jsonschema.validate(result, self.schema) return result except jsonschema.ValidationError as e: prompt f\n\n[Schema 校验失败: {e.message}请修正] # 所有重试失败返回空结构 return self._empty_structure() def _build_system_prompt(self) - str: 构建系统提示词 schema_str json.dumps(self.schema, indent2, ensure_asciiFalse) return ( f你必须输出符合以下 JSON Schema 的 JSON 对象 f不要输出任何其他内容不要 Markdown 代码块标记不要解释文字\n fjson\n{schema_str}\n ) def _empty_structure(self) - dict: 生成符合 Schema 的空结构 if self.schema.get(type) object: result {} for key, prop in self.schema.get(properties, {}).items(): if key in self.schema.get(required, []): prop_type prop.get(type) if prop_type string: result[key] elif prop_type array: result[key] [] elif prop_type object: result[key] {} elif prop_type number: result[key] 0 elif prop_type boolean: result[key] False return result return {}四、边界分析与架构权衡约束解码的性能开销基于状态机的约束解码需要在每一步生成时计算允许的 token 集合增加约 10%-20% 的延迟。对于实时交互场景可以考虑使用 vLLM/Outlines 等框架的 GPU 加速约束解码。Schema 复杂度的限制嵌套层级过深或条件逻辑过复杂的 Schema如 oneOf、anyOf、if-then-else可能导致约束解码器状态爆炸。建议将复杂 Schema 拆分为多个简单 Schema分步生成。格式修复的可靠性修复逻辑基于启发式规则对于严重损坏的 JSON 可能修复出合法但语义错误的结果。建议在修复后增加语义校验如检查枚举值、数值范围。大模型的结构化能力差异不同模型对结构化输出的支持程度不同。GPT-4 和 Claude 3.5 的 JSON 生成准确率可达 95% 以上而开源 7B 模型可能只有 70%-80%。对于准确率要求高的场景建议使用支持原生结构化输出的模型 API如 OpenAI 的 Structured Outputs。五、总结大模型结构化输出通过约束解码、格式修复和 Schema 校验三层保障将自由文本转化为可靠数据。落地要点JSON Schema 定义输出格式规范约束解码器在生成过程中约束 token 选择格式修复兜底处理常见格式错误Schema 校验确保语义正确。关键权衡约束解码保证格式但增加延迟格式修复提高容错但可能引入语义错误简单 Schema 可靠但表达能力有限。
