上周线上服务又出幺蛾子：客服机器人的JSON输出突然多了一个尾逗号，下游解析炸了，300个请求全部500。结构化输出是LLM落地的核心能力，但如果只依赖模型"自觉"输出正确格式，迟早翻车。

这篇文章分享一套从Schema定义到自动修复的完整流水线，把结构化输出的失败率从5%降到0.1%以下。

问题的本质

LLM生成文本是概率过程，即使你给了JSON Schema的示例，模型仍然可能：

• 输出多余字段（“我帮你整理了一下"后跟JSON）
• 字段类型错误（数字返回了字符串）
• 枚举值拼错（“pending” 写成 “Pending”）
• JSON格式损坏（漏逗号、多括号）

核心思路：不要信任模型输出，用程序兜底。

三层防御体系

┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │ Prompt层 │───>│ 校验层 │───>│ 修复层 │ │ 结构化约束 │ │ Schema验证 │ │ 自动修复 │ └─────────────┘ └─────────────┘ └─────────────┘

第一层：Prompt工程约束

在System Prompt中强制JSON Schema是最基本的，但很多人写得太简略：

`python

❌ 典型错误：只给了格式说明，没有约束力

system_prompt = “请以JSON格式返回，包含name和age字段”

✅ 正确做法：给出完整的JSON Schema + 强制约束

system_prompt = "”" 你是一个JSON输出助手。严格遵守以下规则：

1. 输出必须是纯JSON，禁止任何解释文字
2. 严格遵守以下JSON Schema
3. 不要添加注释、markdown代码块标记

JSON Schema: { “type”: “object”, “required”: [“name”, “age”, “status”], “properties”: { “name”: {“type”: “string”, “minLength”: 1}, “age”: {“type”: “integer”, “minimum”: 0, “maximum”: 150}, “status”: {“type”: “string”, “enum”: [“active”, “inactive”, “pending”]} }, “additionalProperties”: false } """ `

第二层：Schema校验

用Pydantic做运行时校验，比手动解析JSON强一百倍：

`python from pydantic import BaseModel, Field, field_validator from enum import Enum from typing import Optional import json

class Status(str, Enum): ACTIVE = “active” INACTIVE = “inactive” PENDING = “pending”

class UserProfile(BaseModel): name: str = Field(…, min_length=1, max_length=100) age: int = Field(…, ge=0, le=150) status: Status email: Optional[str] = None

@field_validator("email")
def validate_email(cls, v):
    if v and "@" not in v:
        raise ValueError("邮箱格式无效")
    return v

class OutputValidator: “““LLM输出校验器”””

def validate(self, raw_output: str, model_class: type[BaseModel]):
    result = {
        "valid": False,
        "data": None,
        "error": None,
        "error_type": None,
        "fixable": False,
    }

    # 步骤1：尝试提取JSON（处理模型输出包裹了额外文字的情况）
    json_str = self._extract_json(raw_output)
    if json_str is None:
        result["error"] = "输出中未找到有效JSON"
        result["error_type"] = "no_json"
        return result

    # 步骤2：尝试解析JSON
    try:
        data = json.loads(json_str)
    except json.JSONDecodeError as e:
        result["error"] = f"JSON解析失败: {e}"
        result["error_type"] = "parse_error"
        result["fixable"] = True  # 格式错误可能能修复
        return result

    # 步骤3：Pydantic模型校验
    try:
        validated = model_class(**data)
        result["valid"] = True
        result["data"] = validated
    except Exception as e:
        result["error"] = f"Schema校验失败: {e}"
        result["error_type"] = "schema_error"
        result["fixable"] = True

    return result

def _extract_json(self, text: str) -> str | None:
    """从可能包含额外文字的输出中提取JSON"""
    import re

    # 尝试直接解析整个文本
    try:
        json.loads(text.strip())
        return text.strip()
    except json.JSONDecodeError:
        pass

    # 尝试提取markdown代码块中的JSON
    pattern = r"`(?:json)?\s*([\s\S]*?)`"
    match = re.search(pattern, text)
    if match:
        return match.group(1).strip()

    # 尝试找到最外层的{...}
    pattern = r"\{[\s\S]*\}"
    match = re.search(pattern, text)
    if match:
        return match.group(0).strip()

    return None

`

第三层：自动修复与重试

校验失败不是终点，而是修复的起点：

`python import openai import asyncio from tenacity import retry, stop_after_attempt, wait_exponential

class ReliableStructuredOutput: “““可靠的结构化输出生成器”””

def __init__(self, client: openai.AsyncOpenAI, model: str = "gpt-4o"):
    self.client = client
    self.model = model
    self.validator = OutputValidator()
    self.max_retries = 3

async def generate(
    self,
    user_prompt: str,
    model_class: type[BaseModel],
    system_prompt: str | None = None,
) -> BaseModel:
    schema = model_class.model_json_schema()
    sys_prompt = system_prompt or self._build_system_prompt(schema)

    # 第一次尝试
    response = await self._call_llm(sys_prompt, user_prompt)
    result = self.validator.validate(response, model_class)

    if result["valid"]:
        return result["data"]

    # 自动修复循环
    for attempt in range(self.max_retries):
        if not result["fixable"]:
            break

        # 构造修复请求：把原始输出和错误信息一起发给模型
        fix_prompt = self._build_fix_prompt(
            user_prompt, response, result["error"], schema
        )
        response = await self._call_llm(sys_prompt, fix_prompt)
        result = self.validator.validate(response, model_class)

        if result["valid"]:
            return result["data"]

    # 所有修复都失败了，抛异常
    raise StructuredOutputError(
        f"经过{self.max_retries}次重试仍无法生成有效输出: {result['error']}"
    )

def _build_system_prompt(self, schema: dict) -> str:
    return f"""你是JSON输出助手。严格只输出JSON，不要任何解释。

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

def _build_fix_prompt(
    self, original_prompt: str, bad_output: str, error: str, schema: dict
) -> str:
    return f"""之前的输出有格式错误，请修正。

原始请求：{original_prompt}

你的输出： {bad_output}

错误信息：{error}

请严格按照JSON Schema重新输出，只返回JSON。"""

async def _call_llm(self, system: str, user: str) -> str:
    resp = await self.client.chat.completions.create(
        model=self.model,
        messages=[
            {"role": "system", "content": system},
            {"role": "user", "content": user},
        ],
        temperature=0,  # 降低温度减少随机性
    )
    return resp.choices[0].message.content

`

使用示例

`python async def main(): client = openai.AsyncOpenAI() generator = ReliableStructuredOutput(client)

# 结构化提取
user_info = await generator.generate(
    user_prompt="我叫张三，今年28岁，在北京工作，邮箱是zhangsan@example.com",
    model_class=UserProfile,
)
print(user_info)
# name='张三' age=28 status=<Status.ACTIVE: 'active'> email='zhangsan@example.com'

`

进阶：批量处理的容错设计

生产环境跑批量任务时，不能因为个别失败就中断：

`python from dataclasses import dataclass, field from typing import TypeVar

T = TypeVar(“T”, bound=BaseModel)

@dataclass class BatchResult: total: int success: int failed: int results: list = field(default_factory=list) errors: list = field(default_factory=list)

class BatchStructuredProcessor: “““批量结构化处理器”””

def __init__(self, generator: ReliableStructuredOutput, concurrency: int = 10):
    self.generator = generator
    self.semaphore = asyncio.Semaphore(concurrency)

async def process_batch(
    self,
    prompts: list[str],
    model_class: type[T],
) -> BatchResult:
    tasks = [self._safe_process(p, model_class, idx) for idx, p in enumerate(prompts)]
    raw_results = await asyncio.gather(*tasks)

    successes = [r for r in raw_results if r["success"]]
    failures = [r for r in raw_results if not r["success"]]

    return BatchResult(
        total=len(prompts),
        success=len(successes),
        failed=len(failures),
        results=[r["data"] for r in successes],
        errors=[{"index": r["index"], "error": r["error"]} for r in failures],
    )

async def _safe_process(self, prompt, model_class, index):
    async with self.semaphore:
        try:
            data = await self.generator.generate(prompt, model_class)
            return {"success": True, "data": data, "index": index}
        except Exception as e:
            return {"success": False, "error": str(e), "index": index}

`

监控指标

上线后需要关注这几个指标：

`python from prometheus_client import Counter, Histogram

OUTPUT_ATTEMPTS = Counter( “llm_output_attempts_total”, “结构化输出尝试次数”, [“model”, “schema”, “attempt”], )

OUTPUT_FAILURES = Counter( “llm_output_failures_total”, “结构化输出最终失败次数”, [“model”, “schema”, “error_type”], )

VALIDATION_DURATION = Histogram( “llm_output_validation_seconds”, “校验耗时”, buckets=[0.001, 0.005, 0.01, 0.05, 0.1], ) `

关键指标：

• 首次成功率：第一次就通过校验的比例，低于90%说明Prompt需要优化
• 重试率：需要重试的比例，高于10%要考虑换更强的模型
• 最终失败率：重试后仍失败的比例，必须低于0.5%

总结

结构化输出的可靠性不是靠"模型够强"就能保证的，需要工程手段兜底。核心三板斧：Prompt Schema约束 + Pydantic运行时校验 + 自动修复重试。配上监控告警，线上稳如老狗。