大模型结构化输出与 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 --> F
2.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 result
2.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 None
2.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}",
temperature=0.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, indent=2, ensure_ascii=False)
return (
f"你必须输出符合以下 JSON Schema 的 JSON 对象，"
f"不要输出任何其他内容（不要 Markdown 代码块标记，不要解释文字）：\n"
f"```json\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 可靠但表达能力有限。
AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念，把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起，为开发者提供从开发、训练到部署的一站式体验。
更多推荐
0
0- 0
已为社区贡献80条内容
所有评论(0)