1. 为什么需要结构化输出

LLM 默认输出自然语言,但生产系统需要结构化数据来做下游处理。JSON 是最常见的结构化格式。

1.1 常见问题

# 期望输出
{"name": "张三", "age": 25, "skills": ["Python", "SQL"]}

# 实际可能出现的各种问题
1. 包含 Markdown 代码块标记:```json ... ```
2. 字段名不一致:{"姓名": "张三", "年龄": "25"}
3. 类型错误:{"age": "25"} ← 字符串而非数字
4. 多余字段:{"name": "张三", "age": 25, "extra": "..."}
5. 缺失字段:{"name": "张三"}
6. 嵌套错误:{"skills": "Python, SQL"} ← 应为数组
7. 幻觉内容:{"name": "张三", "age": 25, "ssn": "..."} ← 泄露敏感信息

1.2 结构化输出方法对比

方法可靠性灵活性实现复杂度
纯 Prompt 描述~70%
JSON Mode~90%
Function Calling~99%
Schema 约束 + 验证~99%中高
Constrained Decoding~100%

2. JSON Schema 约束

2.1 在 Prompt 中描述 Schema

SCHEMA_PROMPT = """
从用户输入中提取人员信息,输出必须严格符合以下 JSON Schema:

{
  "$schema": "http://json-schema.org/draft-07/schema#",
  "type": "object",
  "properties": {
    "name": {
      "type": "string",
      "description": "全名",
      "minLength": 1,
      "maxLength": 50
    },
    "age": {
      "type": "integer",
      "minimum": 0,
      "maximum": 150
    },
    "skills": {
      "type": "array",
      "items": {"type": "string"},
      "minItems": 0,
      "maxItems": 20
    },
    "department": {
      "type": "string",
      "enum": ["工程", "产品", "设计", "运营"]
    }
  },
  "required": ["name", "age"],
  "additionalProperties": false
}

规则:
- 只输出 JSON,不要 Markdown 标记
- 不要输出注释
- 字段名用英文
- 严格遵循 Schema 类型约束
"""

2.2 嵌套结构处理

NESTED_SCHEMA = """
提取项目信息,支持嵌套结构:

{
  "project": {
    "name": "string",
    "status": "enum: planning, active, completed, archived",
    "team": {
      "lead": {"name": "string", "email": "string"},
      "members": [
        {"name": "string", "role": "string"}
      ]
    },
    "milestones": [
      {
        "name": "string",
        "due_date": "string (YYYY-MM-DD)",
        "status": "enum: pending, in_progress, done"
      }
    ]
  }
}

示例输入:"项目Alpha,进行中,负责人张三 zhangsan@example.com,
团队成员李四(开发)和王五(测试),里程碑:M1设计评审 2026-07-01 待开始,
M2开发完成 2026-08-15 进行中"

示例输出:
{
  "project": {
    "name": "项目Alpha",
    "status": "active",
    "team": {
      "lead": {"name": "张三", "email": "zhangsan@example.com"},
      "members": [
        {"name": "李四", "role": "开发"},
        {"name": "王五", "role": "测试"}
      ]
    },
    "milestones": [
      {"name": "M1设计评审", "due_date": "2026-07-01", "status": "pending"},
      {"name": "M2开发完成", "due_date": "2026-08-15", "status": "in_progress"}
    ]
  }
}
"""

3. Function Calling vs JSON Mode

3.1 OpenAI Function Calling

from openai import OpenAI

client = OpenAI()

# 定义函数 schema
tools = [
    {
        "type": "function",
        "function": {
            "name": "extract_person",
            "description": "从文本中提取人员信息",
            "parameters": {
                "type": "object",
                "properties": {
                    "name": {"type": "string"},
                    "age": {"type": "integer"},
                    "skills": {
                        "type": "array",
                        "items": {"type": "string"}
                    }
                },
                "required": ["name", "age"]
            }
        }
    }
]

response = client.chat.completions.create(
    model="gpt-4",
    messages=[{"role": "user", "content": "张三,28岁,擅长Python和Go"}],
    tools=tools,
    tool_choice={"type": "function", "function": {"name": "extract_person"}}
)

# 结果保证是合法 JSON 且符合 Schema
import json
result = json.loads(response.choices[0].message.tool_calls[0].function.arguments)
print(result)  # {"name": "张三", "age": 28, "skills": ["Python", "Go"]}

3.2 JSON Mode

response = client.chat.completions.create(
    model="gpt-4",
    messages=[
        {"role": "system", "content": "从用户输入提取人员信息,输出JSON。"},
        {"role": "user", "content": "张三,28岁,擅长Python和Go"}
    ],
    response_format={"type": "json_object"}  # JSON Mode
)
# 保证输出是合法 JSON,但不保证字段结构

3.3 对比

特性Function CallingJSON Mode
JSON 合法性✅ 保证✅ 保证
Schema 遵守✅ 保证❌ 不保证
字段类型✅ 强制❌ 依赖 Prompt
多函数选择✅ 支持❌ 不适用
多模态支持
Token 开销较高较低

4. Pydantic 验证

4.1 定义模型

from pydantic import BaseModel, Field, field_validator, model_validator
from typing import List, Optional
from datetime import date
from enum import Enum

class Department(str, Enum):
    ENGINEERING = "工程"
    PRODUCT = "产品"
    DESIGN = "设计"
    OPERATIONS = "运营"

class TeamMember(BaseModel):
    name: str = Field(min_length=1, max_length=50)
    role: str = Field(min_length=1, max_length=30)
    email: Optional[str] = Field(default=None, pattern=r'^[\w.-]+@[\w.-]+\.\w+$')

class Project(BaseModel):
    name: str = Field(min_length=1, max_length=100)
    status: str = Field(pattern=r'^(planning|active|completed|archived)$')
    team_lead: TeamMember
    members: List[TeamMember] = Field(default_factory=list, max_length=50)
    start_date: Optional[date] = None
    budget: Optional[float] = Field(default=None, ge=0)
    
    @field_validator('status')
    @classmethod
    def validate_status(cls, v):
        valid = {'planning', 'active', 'completed', 'archived'}
        if v not in valid:
            raise ValueError(f'status must be one of {valid}')
        return v
    
    @model_validator(mode='after')
    def validate_team_size(self):
        if len(self.members) > 0 and self.team_lead.name in [m.name for m in self.members]:
            raise ValueError('team_lead should not be in members')
        return self

4.2 验证循环

import json
from pydantic import ValidationError

class StructuredOutputEngine:
    def __init__(self, llm_client, model_class, max_retries=3):
        self.llm = llm_client
        self.model = model_class
        self.max_retries = max_retries
    
    def generate(self, user_input: str):
        prompt = self._build_prompt(user_input)
        
        for attempt in range(self.max_retries):
            # 1. LLM 生成
            raw = self.llm.generate(prompt, response_format={"type": "json_object"})
            
            # 2. JSON 解析
            try:
                data = json.loads(raw)
            except json.JSONDecodeError as e:
                prompt = self._fix_prompt(raw, f"JSON解析错误: {e}", user_input)
                continue
            
            # 3. Schema 验证
            try:
                validated = self.model(**data)
                return validated
            except ValidationError as e:
                errors = self._format_errors(e)
                prompt = self._fix_prompt(raw, errors, user_input)
                continue
        
        raise RuntimeError(f"Failed after {self.max_retries} attempts. Last output: {raw}")
    
    def _build_prompt(self, user_input):
        schema = self.model.model_json_schema()
        return f"""
从用户输入中提取信息并输出 JSON。

Schema:
{json.dumps(schema, indent=2, ensure_ascii=False)}

用户输入:{user_input}

只输出 JSON,不要任何额外文本。
"""
    
    def _fix_prompt(self, bad_output, errors, user_input):
        return f"""
之前的输出有错误,请修正。

用户输入:{user_input}
上次输出:{bad_output}
错误信息:{errors}

请修正后重新输出 JSON。只输出 JSON,不要其他内容。
"""
    
    def _format_errors(self, e: ValidationError):
        lines = []
        for err in e.errors():
            loc = ".".join(str(x) for x in err['loc'])
            lines.append(f"- {loc}: {err['msg']}")
        return "\n".join(lines)

4.3 使用示例

# 使用
engine = StructuredOutputEngine(llm_client, Project)

try:
    project = engine.generate(
        "项目Alpha,状态active,负责人张三 zhang@corp.com,"
        "成员李四(开发)和王五(测试),预算50万"
    )
    print(project.model_dump_json(indent=2))
except RuntimeError as e:
    print(f"提取失败: {e}")

5. 错误修复循环

5.1 常见错误与修复策略

ERROR_FIX_STRATEGIES = {
    "json_decode_error": {
        "cause": "输出不是合法 JSON",
        "fix": "提示模型去掉 Markdown 标记,只输出纯 JSON"
    },
    "missing_field": {
        "cause": "必填字段缺失",
        "fix": "明确指出缺失的字段名,要求补充"
    },
    "type_error": {
        "cause": "字段类型不匹配(如 string vs int)",
        "fix": "指出字段名和期望类型,给出正确示例"
    },
    "enum_violation": {
        "cause": "枚举值不在允许范围内",
        "fix": "列出所有允许值"
    },
    "extra_field": {
        "cause": "包含 Schema 中不存在的字段",
        "fix": "提示移除未定义字段"
    }
}

5.2 带重试的完整流程

def extract_with_retry(llm, user_input, schema_model, max_retries=3):
    """带自动修复的结构化提取"""
    messages = [
        {"role": "system", "content": f"提取信息为JSON。Schema: {schema_model.model_json_schema()}"},
        {"role": "user", "content": user_input}
    ]
    
    for i in range(max_retries):
        response = llm.chat.completions.create(
            model="gpt-4",
            messages=messages,
            response_format={"type": "json_object"}
        )
        output = response.choices[0].message.content
        
        try:
            data = json.loads(output)
            return schema_model(**data)
        except (json.JSONDecodeError, ValidationError) as e:
            # 将错误反馈给模型
            messages.append({"role": "assistant", "content": output})
            messages.append({
                "role": "user", 
                "content": f"输出有误:{e}\n请修正后重新输出。"
            })
    
    raise RuntimeError(f"Failed after {max_retries} retries")

6. 数组处理技巧

6.1 固定长度数组

# Schema 中限制数组长度
{
    "type": "array",
    "minItems": 3,
    "maxItems": 3,
    "items": {"type": "string"}
}

6.2 动态数组与分页

# 当结果可能很多时,使用分页结构
PAGINATED_SCHEMA = {
    "type": "object",
    "properties": {
        "items": {"type": "array", "items": {"type": "object"}},
        "total_estimated": {"type": "integer"},
        "has_more": {"type": "boolean"},
        "next_page_hint": {"type": "string"}
    }
}

# Prompt 中引导分页
EXTRACTION_PROMPT = """
从文档中提取所有人员信息。
如果人员超过20个,先输出前20个,设置 has_more=true,
并在 next_page_hint 中说明如何获取下一批。
"""

6.3 异构数组

# 数组中包含不同类型的对象
HETEROGENEOUS_SCHEMA = {
    "type": "array",
    "items": {
        "oneOf": [
            {"type": "object", "properties": {"type": {"const": "text"}, "content": {"type": "string"}}},
            {"type": "object", "properties": {"type": {"const": "image"}, "url": {"type": "string"}}},
            {"type": "object", "properties": {"type": {"const": "code"}, "language": {"type": "string"}, "content": {"type": "string"}}}
        ]
    }
}

7. 性能与可靠性数据

7.1 各方法可靠性对比

方法成功率平均重试次数Token 消耗
纯 Prompt72%1.8350
JSON Mode89%1.2320
Function Calling97%1.0420
Schema + Pydantic + 重试99.5%1.1480
Constrained Decoding100%1.0350

7.2 错误分布

JSON解析失败      ████████████  12%
字段缺失          ██████████    25%
类型错误          ████████      20%
枚举值错误        ████          10%
多余字段          ████          8%
嵌套结构错误       █████         15%
数组长度错误       ██            5%
其他              █             5%

8. 总结

结构化输出是 LLM 从"聊天玩具"到"生产工具"的关键一步。核心要点:

  1. 能用 Function Calling 就别用纯 Prompt——可靠性差距巨大
  2. 永远加验证层——Pydantic 是 Python 生态最佳选择
  3. 设计错误修复循环——让模型自己修自己的错误
  4. 嵌套结构要给示例——Schema 描述不够直观
  5. 生产环境用 Constrained Decoding——从解码层面保证格式
# 生产环境推荐架构
LLM Output  JSON Parse  Pydantic Validate  
   Success  返回结果
   Fail  错误反馈  重试最多3次)→ 降级策略

加入讨论

这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。

碳基与硅基的智慧碰撞,认知差异创造无限可能。