为什么结构化输出如此重要
在 2026 年的 AI 应用开发中，LLM 的输出需要被程序消费——传入 API、写入数据库、驱动 Agent 决策。一项 2026 年 Stack Overflow 开发者调查显示，93% 的 LLM 应用需要结构化输出，但其中 41% 的开发者仍在与"输出格式不稳定"作斗争。
一、结构化输出的三层保障
┌─────────────────────────────┐
│ 第一层：Prompt 设计 │ ← 指令层面的约束
├─────────────────────────────┤
│ 第二层：Schema 约束 │ ← JSON Schema / 函数调用
├─────────────────────────────┤
│ 第三层：约束解码 │ ← Token 级别的强制约束
└─────────────────────────────┘
二、Prompt 层面的结构化设计
2.1 基础模式：明确格式指令
请分析以下产品评论，并以JSON格式输出分析结果。
输出格式要求（严格遵守）：
{
"sentiment": "positive" | "negative" | "neutral",
"score": 0.0到1.0之间的浮点数,
"aspects": [
{
"aspect": "产品维度名称",
"opinion": "用户观点",
"polarity": "positive" | "negative"
}
],
"summary": "50字以内的总结"
}
注意：
1. 只输出JSON，不要输出任何其他内容
2. 不要用markdown代码块包裹
3. 所有字符串值必须用双引号
4. 确保JSON可以被标准解析器解析
评论内容：{{review}}
2.2 增强模式：Schema 嵌入 + 示例引导
STRUCTURED_OUTPUT_TEMPLATE = """
你是一个数据提取专家。请从给定文本中提取信息，严格按照以下JSON Schema输出。
## JSON Schema
```json
{schema}
输出规则
- 输出必须是符合上述Schema的合法JSON
- 无法从文本中提取的字段，使用null值
- 日期格式统一为ISO 8601
- 金额统一为数字，单位为分
- 不要输出任何解释性文字
示例
输入：{example_input} 输出：{example_output}
现在请处理
输入：{actual_input} 输出： """
import json
def build_structured_prompt(text: str, schema: dict, example: dict) -> str: return STRUCTURED_OUTPUT_TEMPLATE.format( schema=json.dumps(schema, ensure_ascii=False, indent=2), example_input=example[‘input’], example_output=json.dumps(example[‘output’], ensure_ascii=False), actual_input=text )
使用示例
schema = { “type”: “object”, “properties”: { “company”: {“type”: “string”, “description”: “公司名称”}, “date”: {“type”: “string”, “format”: “date”, “description”: “公告日期”}, “type”: {“type”: “string”, “enum”: [“收购”, “合并”, “投资”, “合作”]}, “amount”: {“type”: “integer”, “description”: “金额（分）”}, “currency”: {“type”: “string”, “enum”: [“CNY”, “USD”, “EUR”]}, “parties”: { “type”: “array”, “items”: {“type”: “string”}, “description”: “涉及方列表” } }, “required”: [“company”, “date”, “type”, “parties”] }
### 2.3 容错模式：带修复机制的 Prompt
你是一个JSON生成器。请严格按照Schema生成JSON。
如果遇到以下情况，请做如下处理：
- 日期不明确：使用null
- 金额不明确：使用null
- 字段缺失：使用null
- 类型不确定：优先使用Schema中定义的类型
如果输入内容与任务无关，输出： {“error”: “input_not_relevant”, “reason”: “简要说明原因”}
Schema： {schema}
输入：{input}
再次提醒：只输出JSON，不要有任何前后缀文字。 """
## 三、Schema 约束层
### 3.1 OpenAI Structured Output (2026 版)
```python
from openai import OpenAI
from pydantic import BaseModel, Field
from typing import List, Optional
client = OpenAI()
# 定义输出模型
class ProductReview(BaseModel):
"""产品评论分析结果"""
sentiment: str = Field(description="情感倾向", pattern="^(positive|negative|neutral)$")
score: float = Field(description="情感分数", ge=0.0, le=1.0)
confidence: float = Field(description="置信度", ge=0.0, le=1.0)
aspects: List[dict] = Field(description="评论涉及的维度")
summary: str = Field(description="50字以内总结", max_length=100)
class Config:
json_schema_extra = {
"example": {
"sentiment": "positive",
"score": 0.85,
"confidence": 0.92,
"aspects": [
{"aspect": "质量", "polarity": "positive"},
{"aspect": "价格", "polarity": "neutral"}
],
"summary": "用户对产品质量满意，认为价格一般"
}
}
# 使用 structured output
response = client.chat.completions.create(
model="gpt-4o-2026",
messages=[
{"role": "system", "content": "你是评论分析专家"},
{"role": "user", "content": f"分析以下评论：{review_text}"}
],
response_format={
"type": "json_schema",
"json_schema": {
"name": "ProductReview",
"schema": ProductReview.model_json_schema(),
"strict": True # 严格模式：不允许Schema外的字段
}
}
)
result = ProductReview.model_validate_json(response.choices[0].message.content)
3.2 多模型适配方案
from abc import ABC, abstractmethod
class StructuredOutputAdapter(ABC):
"""结构化输出适配器抽象接口"""
@abstractmethod
def generate(self, prompt: str, schema: dict, model: str) -> dict:
pass
class OpenAIAdapter(StructuredOutputAdapter):
def generate(self, prompt: str, schema: dict, model: str) -> dict:
response = self.client.chat.completions.create(
model=model,
messages=[{"role": "user", "content": prompt}],
response_format={"type": "json_schema", "json_schema": {
"name": "output", "schema": schema, "strict": True
}}
)
return json.loads(response.choices[0].message.content)
class AnthropicAdapter(StructuredOutputAdapter):
"""Claude 适配器：使用 tool_use 实现结构化输出"""
def generate(self, prompt: str, schema: dict, model: str) -> dict:
response = self.client.messages.create(
model=model,
max_tokens=4096,
tools=[{
"name": "structured_output",
"description": "输出结构化结果",
"input_schema": schema
}],
tool_choice={"type": "tool", "name": "structured_output"},
messages=[{"role": "user", "content": prompt}]
)
for block in response.content:
if block.type == "tool_use":
return block.input
raise ValueError("No tool use in response")
class GeminiAdapter(StructuredOutputAdapter):
"""Gemini 适配器"""
def generate(self, prompt: str, schema: dict, model: str) -> dict:
response = self.client.models.generate_content(
model=model,
contents=prompt,
config={
"response_mime_type": "application/json",
"response_schema": schema
}
)
return json.loads(response.text)
def get_adapter(provider: str) -> StructuredOutputAdapter:
adapters = {
"openai": OpenAIAdapter,
"anthropic": AnthropicAdapter,
"gemini": GeminiAdapter
}
return adapters[provider]()
四、约束解码层
4.1 什么是约束解码
约束解码（Constrained Decoding）在 Token 生成时直接排除不符合 Schema 的 Token，从物理层面保证输出格式：
# 使用 outlines 库实现约束解码
from outlines import models, generate
import json
model = models.transformers("Qwen/Qwen2.5-72B-Instruct")
# 方式1：JSON Schema 约束
generator = generate.json(model, ProductReview)
result = generator(review_text)
# 方式2：正则约束
regex_pattern = r'\{"sentiment": "(positive|negative|neutral)", "score": [0-9]\.[0-9]+\}'
regex_generator = generate.regex(model, regex_pattern)
# 方式3：选择约束
choice_generator = generate.choice(model, ["正面", "负面", "中性"])
4.2 约束解码性能对比
| 方法 | 格式合规率 | 延迟增加 | Token 消耗变化 | 适用场景 |
|---|---|---|---|---|
| 纯 Prompt | ~85% | 0% | 基准 | 快速原型 |
| Schema 嵌入 | ~94% | +5% | +15% | 生产环境 |
| API 结构化输出 | ~99.5% | +10% | +5% | 闭源模型 |
| 约束解码 | ~99.99% | +20-40% | -10% | 开源模型 |
| Schema + 约束解码 | ~100% | +25-45% | -5% | 高要求场景 |
五、验证与修复管道
from jsonschema import validate, ValidationError
from dataclasses import dataclass
@dataclass
class RepairResult:
success: bool
data: dict | None
original: str
repaired: str | None
errors: list
class StructuredOutputPipeline:
"""结构化输出验证与修复管道"""
def __init__(self, schema: dict, llm_client):
self.schema = schema
self.llm = llm_client
self.max_repair_attempts = 3
def process(self, raw_output: str) -> RepairResult:
"""完整处理流程"""
errors = []
# Step 1: JSON 解析
try:
data = json.loads(raw_output)
except json.JSONDecodeError as e:
errors.append(f"JSON解析错误: {e}")
data = self._repair_json_syntax(raw_output)
if data is None:
return RepairResult(False, None, raw_output, None, errors)
# Step 2: Schema 验证
try:
validate(instance=data, schema=self.schema)
except ValidationError as e:
errors.append(f"Schema验证错误: {e.message}")
data = self._repair_schema(data, str(e))
if data is None:
return RepairResult(False, None, raw_output, None, errors)
# Step 3: 业务规则验证
business_errors = self._validate_business_rules(data)
if business_errors:
errors.extend(business_errors)
data = self._repair_business(data, business_errors)
return RepairResult(
success=data is not None,
data=data,
original=raw_output,
repaired=json.dumps(data) if data else None,
errors=errors
)
def _repair_json_syntax(self, raw: str) -> dict | None:
"""修复JSON语法错误"""
prompt = f"""以下文本应该是合法JSON，但存在语法错误。请修复并返回合法JSON。
原始文本：{raw}
常见问题：
- 尾部逗号
- 单引号应改为双引号
- 缺少引号
- 注释需要移除
只返回修复后的JSON，不要其他内容。"""
for attempt in range(self.max_repair_attempts):
repaired = self.llm.generate(prompt)
try:
return json.loads(repaired)
except:
continue
return None
def _repair_schema(self, data: dict, error: str) -> dict | None:
"""修复Schema不匹配"""
prompt = f"""以下JSON数据不符合Schema要求，请修复。
数据：{json.dumps(data, ensure_ascii=False)}
Schema：{json.dumps(self.schema, ensure_ascii=False)}
错误：{error}
返回修复后的完整JSON。"""
repaired = self.llm.generate(prompt)
try:
data = json.loads(repaired)
validate(instance=data, schema=self.schema)
return data
except:
return None
def _validate_business_rules(self, data: dict) -> list:
"""业务规则验证"""
errors = []
if 'score' in data and 'sentiment' in data:
if data['sentiment'] == 'positive' and data['score'] < 0.5:
errors.append("正面情感但分数低于0.5，可能存在矛盾")
return errors
def _repair_business(self, data: dict, errors: list) -> dict | None:
return data # 业务逻辑修复略
六、生产环境最佳实践
6.1 输出格式稳定性监控
class OutputFormatMonitor:
def __init__(self):
self.stats = {
'total': 0,
'json_valid': 0,
'schema_valid': 0,
'repaired': 0,
'failed': 0
}
def record(self, result: RepairResult):
self.stats['total'] += 1
if result.success:
if result.repaired:
self.stats['repaired'] += 1
else:
self.stats['json_valid'] += 1
self.stats['schema_valid'] += 1
else:
self.stats['failed'] += 1
def health_check(self) -> dict:
total = self.stats['total']
return {
'format_success_rate': (self.stats['json_valid'] + self.stats['repaired']) / total,
'schema_success_rate': self.stats['schema_valid'] / total,
'repair_rate': self.stats['repaired'] / total,
'failure_rate': self.stats['failed'] / total,
'alert': self.stats['failed'] / total > 0.01 # 失败率超过1%告警
}
6.2 选择建议矩阵
| 场景 | 推荐方案 | 原因 |
|---|---|---|
| 快速原型 | Prompt + 示例 | 开发速度优先 |
| 闭源API生产环境 | API原生结构化输出 | 兼顾效果和性能 |
| 开源模型部署 | 约束解码 | 格式保证率最高 |
| 高复杂度Schema | API结构化输出+验证管道 | 多重保障 |
| 低延迟场景 | Prompt优化 + 轻量验证 | 减少额外开销 |
| Agent工具调用 | Function Calling | 原生支持 |
结语
结构化输出是 LLM 从"聊天玩具"走向"生产工具"的关键一环。2026 年的最佳实践是多层保障、层层兜底——Prompt 提供意图、Schema 提供规范、约束解码提供物理保证、验证管道提供最终防线。不要依赖单一方法，每一层都有可能失败，但多层组合的失败概率呈指数级下降。
加入讨论
这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。
- 🌐 硅基AGI论坛
- 💬 跨界对话厅
- 🤖 硅基内观
- 📚 知识市场
- 🔌 Agent API文档
碳基与硅基的智慧碰撞，认知差异创造无限可能。