LLM结构化输出与JSON模式工程化深度实战：从Schema约束到函数调用与Pydantic验证的全解析

引言

LLM默认输出自由文本，但企业应用需要结构化数据（JSON/XML/表格）。结构化输出的可靠性是生产系统的硬需求——一个格式错误的JSON可能导致整个流水线崩溃。本文从结构化输出的技术手段讲起，覆盖JSON Mode与Structured Outputs（OpenAI）、Pydantic模型约束、函数调用作为结构化输出通道、输出解析与重试、正则约束生成、XML标签输出模式、Schema版本管理、测试与回归，构建LLM结构化输出的工程体系。

一、结构化输出方法对比

# structured/methods.py
STRUCTURED_OUTPUT_METHODS = {
    "json_mode": {
        "description": "OpenAI JSON Mode：response_format={type:'json_object'}",
        "guarantee": "输出是合法JSON，但不保证schema",
        "best_for": "简单JSON输出",
        "limitation": "需要prompt中描述期望格式",
    },
    "structured_output": {
        "description": "OpenAI Structured Outputs：response_format={type:'json_schema',json_schema:{...}}",
        "guarantee": "输出严格符合JSON Schema",
        "best_for": "需要严格schema保证的生产场景",
        "limitation": "Schema复杂度有限制",
    },
    "function_calling": {
        "description": "通过函数调用参数获得结构化输出",
        "guarantee": "参数符合函数定义的JSON Schema",
        "best_for": "结构化输出+工具执行",
        "limitation": "需要定义函数",
    },
    "pydantic_prompt": {
        "description": "在prompt中描述Pydantic模型，后解析验证",
        "guarantee": "解析时验证，不保证模型输出合法",
        "best_for": "多模型兼容（非OpenAI）",
        "limitation": "需要解析+重试逻辑",
    },
    "regex_constraint": {
        "description": "推理引擎层面用正则约束token生成",
        "guarantee": "输出严格匹配正则",
        "best_for": "vLLM/SGLang等推理引擎",
        "limitation": "正则复杂度有限",
    },
}

二、Pydantic模型约束

# structured/pydantic_output.py
from pydantic import BaseModel, Field, ValidationError
from typing import Optional, Any
import json

class StructuredOutputParser:
    """Pydantic驱动的结构化输出解析器"""
    
    @staticmethod
    def build_prompt(instruction: str, model_class: type[BaseModel]) -> str:
        """构建带Schema约束的提示词"""
        schema = model_class.model_json_schema()
        properties = schema.get("properties", {})
        fields_desc = "\n".join(
            f"- {name}: {prop.get('type', 'any')} - {prop.get('description', '')}"
            for name, prop in properties.items()
        )
        required = schema.get("required", [])
        return f"""{instruction}

请严格按照以下JSON Schema输出，不要包含其他内容：
{json.dumps(schema, ensure_ascii=False, indent=2)}

字段说明：
{fields_desc}

必填字段：{', '.join(required)}

输出JSON："""
    
    @staticmethod
    async def parse_and_validate(raw_output: str,
                                  model_class: type[BaseModel],
                                  llm_client=None,
                                  max_retries: int = 2) -> BaseModel:
        """解析并验证输出，失败时自动重试"""
        # 提取JSON（处理markdown代码块包裹）
        json_str = StructuredOutputParser._extract_json(raw_output)
        
        for attempt in range(max_retries + 1):
            try:
                data = json.loads(json_str)
                return model_class.model_validate(data)
            except (json.JSONDecodeError, ValidationError) as e:
                if attempt < max_retries and llm_client:
                    # 让LLM修复格式
                    fix_prompt = f"""以下JSON解析失败，请修复格式。

原始输出：
{raw_output[:1000]}

错误：
{str(e)[:500]}

期望Schema：
{json.dumps(model_class.model_json_schema(), ensure_ascii=False)[:1000]}

请输出修复后的合法JSON："""
                    raw_output = await llm_client.complete(
                        fix_prompt, temperature=0, max_tokens=2000,
                    )
                    json_str = StructuredOutputParser._extract_json(raw_output)
                else:
                    raise
        
    @staticmethod
    def _extract_json(text: str) -> str:
        """从文本中提取JSON（处理markdown代码块）"""
        import re
        # 尝试从```json代码块提取
        match = re.search(r'```(?:json)?\s*\n?(.*?)\n?```', text, re.S)
        if match:
            return match.group(1).strip()
        # 尝试直接找JSON对象
        match = re.search(r'\{.*\}', text, re.S)
        if match:
            return match.group(0)
        return text.strip()

# 示例模型定义
class ProductExtraction(BaseModel):
    """商品信息提取模型"""
    name: str = Field(description="商品名称")
    price: float = Field(description="价格（人民币）", ge=0)
    currency: str = Field(default="CNY", description="货币代码")
    category: str = Field(description="商品类别")
    attributes: dict[str, str] = Field(
        default_factory=dict, description="商品属性键值对"
    )
    in_stock: bool = Field(default=True, description="是否有库存")
    rating: Optional[float] = Field(
        default=None, description="评分0-5", ge=0, le=5
    )

# 使用
prompt = StructuredOutputParser.build_prompt(
    "从以下商品描述中提取结构化信息：iPhone 15 Pro 256GB钛原色，售价8999元...",
    ProductExtraction,
)
raw = await llm.complete(prompt, temperature=0)
product = await StructuredOutputParser.parse_and_validate(raw, ProductExtraction, llm)
print(product.name)  # "iPhone 15 Pro"
print(product.price)  # 8999.0

三、OpenAI Structured Outputs

# structured/openai_structured.py
from openai import AsyncOpenAI

class OpenAIStructuredOutput:
    """OpenAI Structured Outputs：严格Schema保证"""
    
    def __init__(self, api_key: str):
        self.client = AsyncOpenAI(api_key=api_key)
    
    async def extract(self, text: str, schema: dict,
                      model: str = "gpt-4o-2024-08-06") -> dict:
        """使用JSON Schema约束输出"""
        response = await self.client.chat.completions.create(
            model=model,
            messages=[
                {"role": "system", "content": "从文本中提取结构化信息。"},
                {"role": "user", "content": text},
            ],
            response_format={
                "type": "json_schema",
                "json_schema": {
                    "name": "extraction",
                    "strict": True,  # 严格模式
                    "schema": schema,
                },
            },
            temperature=0,
        )
        import json
        return json.loads(response.choices[0].message.content)
    
    async def classify(self, text: str, categories: list[str]) -> dict:
        """文本分类：结构化输出"""
        schema = {
            "type": "object",
            "properties": {
                "category": {
                    "type": "string",
                    "enum": categories,
                },
                "confidence": {"type": "number"},
                "reasoning": {"type": "string"},
            },
            "required": ["category", "confidence", "reasoning"],
            "additionalProperties": False,
        }
        return await self.extract(
            f"分类以下文本：{text}", schema,
        )

四、批量结构化提取

# structured/batch.py
import asyncio
from typing import Any

class BatchStructuredExtractor:
    """批量结构化提取：并行处理+错误隔离"""
    
    def __init__(self, llm_client, model_class: type[BaseModel],
                 max_concurrency: int = 10):
        self.llm = llm_client
        self.model_class = model_class
        self.semaphore = asyncio.Semaphore(max_concurrency)
    
    async def extract_batch(self, texts: list[str]) -> list[dict]:
        """批量提取"""
        tasks = [self._extract_one(text) for text in texts]
        results = await asyncio.gather(*tasks, return_exceptions=True)
        
        output = []
        for text, result in zip(texts, results):
            if isinstance(result, Exception):
                output.append({
                    "success": False,
                    "error": str(result)[:200],
                    "input": text[:100],
                })
            else:
                output.append({
                    "success": True,
                    "data": result.model_dump(),
                })
        return output
    
    async def _extract_one(self, text: str) -> BaseModel:
        async with self.semaphore:
            prompt = StructuredOutputParser.build_prompt(
                f"从以下文本提取信息：\n{text[:2000]}",
                self.model_class,
            )
            raw = await self.llm.complete(prompt, temperature=0, max_tokens=1000)
            return await StructuredOutputParser.parse_and_validate(
                raw, self.model_class, self.llm,
            )
    
    def summary(self, results: list[dict]) -> dict:
        success = sum(1 for r in results if r["success"])
        return {
            "total": len(results),
            "successful": success,
            "failed": len(results) - success,
            "success_rate": success / max(len(results), 1),
        }

五、回归测试

# structured/testing.py
class StructuredOutputTestSuite:
    """结构化输出回归测试"""
    
    def __init__(self, llm_client):
        self.llm = llm_client
        self.cases: list[dict] = []
    
    def add_case(self, input: str, expected_keys: list[str],
                 model_class: type[BaseModel]):
        self.cases.append({
            "input": input, "expected_keys": expected_keys,
            "model_class": model_class,
        })
    
    async def run(self) -> dict:
        results = []
        for case in self.cases:
            prompt = StructuredOutputParser.build_prompt(
                case["input"], case["model_class"],
            )
            raw = await self.llm.complete(prompt, temperature=0)
            try:
                parsed = await StructuredOutputParser.parse_and_validate(
                    raw, case["model_class"],
                )
                # 验证字段
                data = parsed.model_dump()
                missing = [k for k in case["expected_keys"] if k not in data]
                results.append({
                    "input": case["input"][:100],
                    "success": len(missing) == 0,
                    "missing_keys": missing,
                })
            except Exception as e:
                results.append({
                    "input": case["input"][:100],
                    "success": False,
                    "error": str(e)[:200],
                })
        
        passed = sum(1 for r in results if r["success"])
        return {
            "total": len(results),
            "passed": passed,
            "failed": len(results) - passed,
            "pass_rate": passed / max(len(results), 1),
            "details": results,
        }

总结

LLM结构化输出的工程体系以"约束-解析-验证-重试-测试"五层展开：OpenAI Structured Outputs以JSON Schema+strict模式提供最强保证（输出100%符合Schema），Pydantic模型约束在Prompt中描述Schema并在解析时验证（兼容所有LLM），函数调用以参数Schema为结构化输出通道，输出解析器从markdown代码块或裸JSON中提取并自动修复格式错误（最多2轮LLM修复），批量提取以信号量控制并发并隔离单个失败，回归测试套件验证字段完整性与解析成功率。当LLM输出从"自由文本"变为"可验证的结构化数据"，下游的数据库写入、API响应、报表生成才能可靠自动化，这是LLM从"文本生成器"走向"数据处理引擎"的关键工程能力。