LLM 能力集成：结构化输出与 JSON Schema 约束的工程实践

LLM 能力集成：结构化输出与 JSON Schema 约束的工程实践

一、不确定性之困：大模型输出的"格式失控"

大语言模型本质上是一个概率性的文本生成器。当应用需要模型返回结构化数据（如 JSON 对象、表格行、API 参数）时，这种概率性就变成了工程噩梦。模型可能返回格式正确的 JSON，也可能在 JSON 中混入解释性文字、使用不一致的字段名、甚至生成语法错误的 JSON 字符串。

生产环境中，这种"格式失控"的后果是严重的：下游解析器崩溃、数据入库失败、自动化流程中断。更棘手的是，格式错误的出现概率与 Prompt 设计、模型版本、输入长度等因素高度相关，难以稳定复现和修复。

结构化输出（Structured Output）是解决这一问题的工程方案。它通过 JSON Schema 约束模型的输出格式，将"希望模型返回合法 JSON"的软约束升级为"模型必须返回合法 JSON"的硬保证。本文将从约束机制、工程实现和可靠性保障三个层面，深入探讨结构化输出的生产级实践。

二、约束机制：从 Prompt 约束到 Token 级强制

结构化输出的实现经历了三个阶段的演进，每个阶段的可靠性逐步提升。

flowchart LR subgraph S1["阶段一：Prompt 约束"] A1[在 Prompt 中声明<br/>输出格式要求] --> A2[模型"尽力"遵守<br/>无硬性保证] A2 --> A3[后处理解析<br/>容错与重试] end subgraph S2["阶段二：Grammar 约束"] B1[定义 Context-Free Grammar<br/>或 JSON Schema] --> B2[解码时约束 Token 采样<br/>只允许合法 Token] B2 --> B3[输出 100% 语法合法<br/>但语义可能不合理] end subgraph S3["阶段三：Token 级强制"] C1[JSON Schema 编译为<br/>约束状态机] --> C2[每个生成步骤<br/>动态计算合法 Token 集合] C2 --> C3[输出 100% 格式合法<br/>且语义更合理] end S1 -->|可靠性提升| S2 S2 -->|语义质量提升| S3

阶段一：Prompt 约束。在系统提示中明确要求模型输出 JSON 格式，并提供示例。这是最简单的方案，但可靠性最低——模型可能在 JSON 外添加解释文字，或生成不符合 Schema 的字段。

阶段二：Grammar 约束。通过约束解码（Constrained Decoding），在每一步 Token 生成时，只允许选择语法合法的 Token。例如，当已生成{"name": "时，下一个 Token 必须是合法的字符串内容或闭合引号，不可能生成}或数字。这保证了输出的语法合法性，但无法保证语义正确性（如字段值类型错误）。

阶段三：Token 级强制。将 JSON Schema 编译为约束状态机，在解码的每一步动态计算合法 Token 集合。这不仅能保证语法合法，还能保证输出符合 Schema 定义的类型、枚举值、数值范围等约束。OpenAI 的 Structured Outputs 功能和 Instructor 库都采用了这一方案。

三、工程实现：生产级结构化输出的完整方案

3.1 基于 Pydantic 的 Schema 定义与验证

# schemas.py — 结构化输出的数据模型定义 from pydantic import BaseModel, Field, field_validator from typing import Literal, Optional from enum import Enum class SentimentType(str, Enum): POSITIVE = "positive" NEGATIVE = "negative" NEUTRAL = "neutral" class Entity(BaseModel): """实体提取结果""" name: str = Field(description="实体名称") entity_type: Literal["person", "organization", "location", "product"] = Field( description="实体类型" ) confidence: float = Field( description="置信度，0-1之间", ge=0.0, le=1.0 ) class AnalysisResult(BaseModel): """文本分析的结构化输出""" summary: str = Field( description="文本摘要，不超过200字", max_length=200 ) sentiment: SentimentType = Field(description="情感倾向") entities: list[Entity] = Field( description="提取的实体列表", min_length=0, max_length=10 ) key_topics: list[str] = Field( description="关键主题列表", min_length=1, max_length=5 ) action_items: Optional[list[str]] = Field( default=None, description="待办事项列表，如无则为null" ) @field_validator("key_topics") @classmethod def validate_topics_not_empty(cls, v): """确保每个主题非空且不重复""" cleaned = [t.strip() for t in v if t.strip()] if len(cleaned) != len(set(cleaned)): raise ValueError("主题列表中存在重复项") return cleaned

3.2 多层保障的结构化输出调用

# structured_output_client.py import json import logging from typing import Type, TypeVar from pydantic import BaseModel, ValidationError from openai import OpenAI logger = logging.getLogger("structured-output") T = TypeVar("T", bound=BaseModel) class StructuredOutputClient: """ 结构化输出客户端：多层保障策略 设计考量：即使模型支持原生 Structured Outputs，仍需后验证兜底 """ def __init__(self, client: OpenAI, model: str = "gpt-4o"): self.client = client self.model = model self._retry_config = { "max_retries": 3, "retry_delay": 1.0, # 秒 } def generate( self, prompt: str, response_model: Type[T], system_prompt: str | None = None, ) -> T: """ 生成结构化输出，包含三层保障： 1. 原生 Structured Outputs（如果模型支持） 2. JSON Mode 回退 3. Pydantic 后验证 + 自动修复 """ messages = [] if system_prompt: messages.append({"role": "system", "content": system_prompt}) messages.append({"role": "user", "content": prompt}) # 第一层：尝试原生 Structured Outputs try: result = self._call_with_structured_output(messages, response_model) logger.info("Structured Output 原生模式成功") return result except Exception as e: logger.warning(f"Structured Output 原生模式失败: {e}") # 第二层：回退到 JSON Mode + 后验证 for attempt in range(self._retry_config["max_retries"]): try: result = self._call_with_json_mode(messages, response_model) logger.info(f"JSON Mode 回退成功，第 {attempt + 1} 次尝试") return result except (json.JSONDecodeError, ValidationError) as e: logger.warning(f"JSON Mode 第 {attempt + 1} 次失败: {e}") # 在重试时追加纠错提示 messages.append({ "role": "assistant", "content": "格式错误，请重试" }) messages.append({ "role": "user", "content": f"上一次输出格式有误：{str(e)}。请严格按照 JSON Schema 输出。" }) raise RuntimeError("结构化输出生成失败，已耗尽重试次数") def _call_with_structured_output(self, messages: list, response_model: Type[T]) -> T: """使用模型原生 Structured Outputs 能力""" response = self.client.chat.completions.create( model=self.model, messages=messages, response_format={ "type": "json_schema", "json_schema": { "name": response_model.__name__, "schema": response_model.model_json_schema(), "strict": True # 启用严格模式 } } ) raw = json.loads(response.choices[0].message.content) return response_model.model_validate(raw) def _call_with_json_mode(self, messages: list, response_model: Type[T]) -> T: """回退方案：JSON Mode + Pydantic 验证""" # 在 Prompt 中追加 Schema 描述 schema_prompt = ( f"\n\n请严格按照以下 JSON Schema 输出，不要添加任何额外文字：\n" f"```json\n{json.dumps(response_model.model_json_schema(), ensure_ascii=False, indent=2)}\n```" ) messages[-1]["content"] += schema_prompt response = self.client.chat.completions.create( model=self.model, messages=messages, response_format={"type": "json_object"}, temperature=0.1 # 低温度减少格式变异 ) raw = json.loads(response.choices[0].message.content) return response_model.model_validate(raw)

3.3 流式结构化输出的增量解析

# streaming_parser.py import json from typing import Iterator class IncrementalJSONParser: """ 增量 JSON 解析器：在流式输出过程中实时提取已完成的字段 设计考量：避免等待完整 JSON 生成后才解析，提升用户体验 """ def __init__(self): self._buffer = "" self._depth = 0 self._in_string = False self._escape_next = False def feed(self, chunk: str) -> dict | None: """输入一个 Token 片段，返回已完成的顶层字段（如果有）""" self._buffer += chunk # 尝试解析当前缓冲区为完整 JSON try: result = json.loads(self._buffer) self._buffer = "" return result except json.JSONDecodeError: pass # 尝试提取已完成的顶层字段 # 简化实现：检测完整的 "key": value 对 return None def parse_stream(self, token_stream: Iterator[str]) -> Iterator[dict]: """消费流式 Token，在 JSON 完成时 yield 解析结果""" for token in token_stream: result = self.feed(token) if result is not None: yield result

四、可靠性的代价：结构化输出的 Trade-offs

4.1 输出质量下降

强制结构化约束会压缩模型的"自由度"。当模型被限制在严格的 Schema 内时，可能牺牲内容的丰富性和准确性。例如，要求模型将情感分类为三个固定枚举值时，模型可能被迫选择一个不精确的类别，而非用自然语言描述微妙的情感。

4.2 Token 消耗增加

JSON Schema 的结构标记（键名、括号、引号）会占用额外的 Token。在复杂 Schema 场景下，结构化输出的 Token 消耗可能比自由文本输出增加 20-40%，直接推高 API 调用成本。

4.3 延迟增加

约束解码需要在每一步计算合法 Token 集合，这增加了推理延迟。在长输出场景下，结构化输出的生成速度可能比自由文本慢 15-30%。

4.4 Schema 设计的约束

并非所有 JSON Schema 特性都被模型支持。例如，additionalProperties: false在某些模型实现中可能导致意外行为；嵌套过深的 Schema（超过 5 层）可能超出模型的上下文理解能力。

4.5 适用边界

结构化输出最适合：API 参数提取、表单数据填充、分类与标注任务、知识图谱三元组提取等对格式要求严格的场景。不适合：创意写作、开放式问答、需要模型自由发挥的场景。

五、总结

结构化输出将大模型从"文本生成器"升级为"数据生成器"，是 LLM 工程化落地的关键基础设施。从 Prompt 约束到 Token 级强制，可靠性逐步提升，但代价是输出质量、Token 消耗和延迟的权衡。工程实践中的核心策略是"多层保障"：优先使用原生 Structured Outputs，回退到 JSON Mode + Pydantic 验证，流式场景采用增量解析。Schema 设计应遵循"最小约束"原则——只约束必须约束的字段，给模型留出合理的表达空间。理解约束与自由度之间的张力，是构建可靠 LLM 应用的基本功。