结构化输出与 schema 校验：从模型到业务的数据流工程

2026-07-03

结构化输出不只是”让模型输出 JSON”，而是建立从 prompt → 模型 → schema 校验 → 业务层的完整数据管道，保证每个环节的类型安全和错误可恢复。JSON 合法是起点，业务 schema 校验才是终点。

如果你写过接模型输出入库的代码，大概率踩过这个坑：本地测了十几条 prompt 都正常，一上线跑批量数据就开始报 json.decoder.JSONDecodeError: Expecting value: line 1 column 1 (char 0)。翻半天日志发现模型压根没输出 JSON，而是先来了一句”好的，以下是审查结果：“，再跟一段 ```json 包裹的代码块。这不是模型抽风，是你只做了”提示模型输出 JSON”这一层，没有在管道里加校验和兜底。这篇讲的就是怎么把这条链路从”祈祷模型听话”变成”工程上可控”。

Schema 设计原则

好的 schema 设计决定了校验的成功率和维护成本：

from pydantic import BaseModel, Field, field_validator
from typing import Literal
from enum import Enum

class Severity(str, Enum):
    HIGH = "high"
    MEDIUM = "medium"
    LOW = "low"

class Issue(BaseModel):
    line: int = Field(ge=1, description="行号，从1开始")
    severity: Severity
    message: str = Field(min_length=5, max_length=200)
    suggestion: str | None = None

class CodeReviewResult(BaseModel):
    summary: str = Field(max_length=500)
    issues: list[Issue] = Field(default_factory=list)
    score: float = Field(ge=0, le=100)
    pass_review: bool

    @field_validator("issues")
    @classmethod
    def validate_issues_count(cls, v):
        if len(v) > 50:
            raise ValueError("issues 数量不能超过 50 条")
        return v

关键设计决策：

用 Enum 而非 Literal[...] 管理枚举值，方便扩展
Field(ge=0, le=100) 比注释描述更可靠，校验时自动检查
可选字段用 X | None，不要让模型猜”这字段要不要填”

这三条背后各有一段踩坑史，值得展开讲讲。

先说 Enum 和 Literal 的区别。刚开始我也图省事直接写 Literal["high", "medium", "low"]，能跑，但半年后业务方说要加一个 critical 级别，你得去代码里全局搜这个 Literal 出现的每一处改掉——校验函数里改一处，序列化里再改一处，前端类型定义再对一遍。换成 Enum 之后，新增级别只需要改这一个类定义，model_json_schema() 自动把新值同步进 schema，传给 OpenAI 的 JSON Schema 里 enum 数组也是自动生成的，不用手写。对于会随业务迭代的枚举字段，一开始就用 Enum，别嫌麻烦。

再说 Field(ge=0, le=100) 这种范围约束。你可能想，反正 prompt 里已经写了”评分范围 0-100”，还需要 Pydantic 再校验一遍吗？需要，而且是刚需。模型偶尔会输出 105 或者 -3 这种越界值，尤其是在你要求它”严格一点""扣分要狠”这类主观指令之后，模型有时会把百分制当成”越离谱越能体现严格”，直接给出超出范围的分数。Field 里的约束是运行时强校验，一旦越界就抛 ValidationError，你可以接到重试链路里让模型重新打分，而不是让一个 105 分的脏数据流进数据库或者报表里，等运营发现”这个分数怎么比满分还高”的时候已经是两周后的事故复盘会了。

最后说可选字段。很多人图省事把 suggestion 这种非必填字段直接留空字符串默认值 suggestion: str = ""，这样模型不填的时候也不会报错。问题是，你后续没法区分”模型认为不需要给建议”和”模型没生成这个字段”——业务层想统计”多少条 issue 缺失整改建议”就没法做了。用 str | None = None 之后，None 明确代表”模型没给”，空字符串 "" 代表”模型给了但内容为空”，语义是分开的，排查问题的时候能省很多事。

OpenAI Structured Outputs（原生 schema 绑定）

OpenAI gpt-4o-2024-08-06 及以上支持在 API 层绑定 JSON Schema：

from openai import OpenAI
import json

client = OpenAI()

response = client.chat.completions.create(
    model="gpt-4o-2024-08-06",
    messages=[
        {"role": "system", "content": "你是代码审查专家，审查以下代码并按格式输出"},
        {"role": "user", "content": code_content}
    ],
    response_format={
        "type": "json_schema",
        "json_schema": {
            "name": "code_review",
            "strict": True,
            "schema": CodeReviewResult.model_json_schema()  # Pydantic 生成 schema
        }
    }
)

# 模型保证输出符合 schema，直接解析
raw = json.loads(response.choices[0].message.content)
result = CodeReviewResult.model_validate(raw)

strict: True 模式下模型会拒绝输出不符合 schema 的内容，但会增加约 5-10% 的延迟。

这里有几个 strict 模式的真实限制，不看官方文档很容易踩：一是所有字段都会被当成必填对待，就算你在 Pydantic 里写了 X | None = None，schema 层面也得把它标进 required 数组，只是允许值为 null——这跟很多人的直觉（可选字段可以不出现在 JSON 里）是反的，模型一定会把这个 key 输出出来，只是值可能是 null。二是 additionalProperties 必须显式设为 false，Pydantic 的 model_json_schema() 默认生成的 schema 符合这个要求，但如果你手写 schema 又忘了加这一条，strict 模式会直接报参数错误而不是运行时校验错误，调用都发不出去。三是并不是所有 JSON Schema 关键字都支持，比如 minLength/maxLength 这种字符串长度约束在早期版本的 strict 模式下会被忽略（模型不保证遵守），这类约束建议还是放在 Pydantic 里做二次校验，别指望 API 层帮你兜底——这条以官方最新文档为准，具体支持范围可能随版本更新。

对比一下几种”让模型出结构化数据”的路子，选型时能少走弯路：

方案	强约束程度	延迟	适用场景
纯 prompt 要求输出 JSON	弱，靠模型自觉	最低	快速原型、内部工具、容错要求低
Function calling / tool use	中，模型倾向遵守但不保证	略增	需要模型”决定调用哪个函数”的多工具场景
Structured Outputs（`strict: true`）	强，API 层拒绝非法输出	增加约 5-10%	生产环境、下游要直接反序列化入库的场景

如果你的模型不支持原生 Structured Outputs（比如接的是走 Function Calling 路线的模型，或者国内厂商的部分模型只支持宽松的 response_format: {type: "json_object"}），退化方案就是本文后面讲的”重试 + Pydantic 校验”这一套，本质是把 API 层做不到的强约束，挪到你自己的代码里做。

自动重试与错误修复

校验失败时的重试策略：

from pydantic import ValidationError
import time

async def call_with_retry(
    prompt: str,
    response_model: type[BaseModel],
    max_retries: int = 3
) -> BaseModel:
    last_error = None
    
    for attempt in range(max_retries):
        try:
            raw = await call_llm(prompt)
            return response_model.model_validate_json(raw)
        except (ValidationError, json.JSONDecodeError) as e:
            last_error = e
            # 把错误信息注入下一次提示，引导模型自我修正
            prompt = f"""
{prompt}

上次输出校验失败，错误：{str(e)[:200]}
请修正后重新输出合法的 JSON。
"""
            if attempt < max_retries - 1:
                await asyncio.sleep(0.5 * (attempt + 1))  # 指数退避
    
    raise RuntimeError(f"重试 {max_retries} 次后仍然失败: {last_error}")

这段重试逻辑里有几个细节值得展开：

第一，为什么要把错误信息截断到 str(e)[:200] 再注入下一轮 prompt？因为 Pydantic 的 ValidationError 完整信息可能包含好几百字符的嵌套字段路径，全部塞进 prompt 会占用不必要的 token，而且模型往往只需要知道”哪个字段、什么问题”就够了，太长的错误堆栈反而会分散模型的注意力，修正效果没有精简版好。

第二，指数退避 0.5 * (attempt + 1) 是线性退避不是真正的指数退避（真正的指数应该是 0.5 * (2 ** attempt)），这里用线性是因为校验失败大概率是模型”没理解格式要求”，不是限流问题，不需要拉长等待时间去避让频率限制；如果你的重试场景里混了 429（限流）错误，那两种失败原因要分开处理——429 走真正的指数退避加长等待，schema 校验失败走短间隔快速重试，混在一起用同一套退避策略等于该等的没等够、不该等的白等。

第三，这套重试对”模型持续输出错误格式”这类问题有效，但治不了”模型输出内容本身跑题”的问题。比如你要它审查代码里的安全漏洞，它给你返回了合法的 JSON，但内容是”这段代码写得很好，没有问题”（敷衍了事）。这种情况 schema 校验会通过，因为格式没错，但业务上是废的。想抓这类问题，得在 pass_review 之类的字段上加业务规则校验，比如”如果 issues 为空但 score 低于 60，视为异常，触发人工复核”，这已经超出 schema 校验的范畴，是业务层规则引擎的活了。

嵌套 schema 的拆分策略

对于复杂嵌套结构，一次输出全部容易出错。推荐分步生成：

第一步：生成顶层摘要（summary, score, pass_review）
第二步：基于摘要，逐段生成 issues 列表
第三步：合并两步结果，做整体校验

为什么大 schema 一次性生成容易翻车？本质是模型的注意力预算有限。一个包含十几个字段、多层嵌套数组的 schema，模型在生成过程中要同时维护”当前在哪个字段""这个数组还要不要继续追加元素""前面填的内容跟当前字段有没有矛盾”这几件事，字段越多、嵌套越深，出错概率越高——常见翻车现场是数组提前截断（该有 5 条 issue 只给了 2 条就把 JSON 括号闭合了），或者字段类型错位（score 该填数字结果套了个对象进去）。拆成两步之后，第一步的 schema 只有 3 个字段，模型专注力集中，出错率明显下降；第二步单独生成 issues 列表，即便数组长也不用同时兼顾摘要字段的准确性。代价是多了一次 API 调用，延迟和成本都翻倍，所以这个策略不是无脑用在所有场景，字段数在 5 个以内、嵌套不超过两层的 schema，直接一次性生成通常就够稳，没必要为了”更保险”多打一次 API。

async def two_stage_review(code: str) -> CodeReviewResult:
    # 第一步：生成摘要
    summary_raw = await call_llm_structured(
        f"审查以下代码，只输出摘要和评分：\n{code}",
        SummaryOnly  # 简化 schema
    )
    
    # 第二步：生成详细 issues
    issues_raw = await call_llm_structured(
        f"基于以下摘要，列出具体问题：\n摘要：{summary_raw.summary}\n代码：{code}",
        IssueList  # 只含 issues 字段的 schema
    )
    
    # 合并
    return CodeReviewResult(
        summary=summary_raw.summary,
        score=summary_raw.score,
        pass_review=summary_raw.pass_review,
        issues=issues_raw.issues
    )

降级与告警

校验最终失败时，不要让业务崩溃：

from dataclasses import dataclass

@dataclass
class StructuredResult:
    data: BaseModel | None
    raw_text: str
    success: bool
    error: str | None = None

async def safe_structured_call(prompt, model) -> StructuredResult:
    try:
        data = await call_with_retry(prompt, model)
        return StructuredResult(data=data, raw_text="", success=True)
    except Exception as e:
        raw = await call_llm(prompt)  # 降级到普通文本输出
        alert(f"结构化输出失败，已降级: {e}")
        return StructuredResult(data=None, raw_text=raw, success=False, error=str(e))

这段代码的关键点是”降级”和”报错”是两件不同的事，很多人只做了报错，没做降级。校验彻底失败之后，与其让接口直接 500，不如把模型的原始输出原样存下来（raw_text），返回给业务层一个明确的 success=False，让上层决定是走人工兜底、还是展示一个”生成失败，请重试”的提示，而不是让用户看到一个裸的 500 页面。alert() 这一行在真实项目里对接的是企业微信机器人或者 Sentry，报警内容一定要带上原始输出的前 200 字符，不然你排查的时候只知道”失败了”，不知道模型到底吐了什么鬼东西出来。

再说回文章开头那个 Expecting value: line 1 column 1 的报错——根因是模型在 JSON 前面加了自然语言前缀，或者把 JSON 包在 ```json 代码块里。这是没开 strict 模式、纯靠 prompt 约束输出格式时最常见的翻车现场。解析前一定要做一层清洗，别指望 prompt 里写”只输出 JSON，不要有任何其他文字”就能杜绝：

import re

def extract_json(raw: str) -> str:
    # 去掉 markdown 代码块包裹
    match = re.search(r"```(?:json)?\s*([\s\S]*?)```", raw)
    if match:
        return match.group(1).strip()
    # 兜底：截取第一个 { 到最后一个 } 之间的内容
    start, end = raw.find("{"), raw.rfind("}")
    if start != -1 and end != -1:
        return raw[start:end + 1]
    return raw.strip()

这层清洗建议放在 call_with_retry 解析之前统一做，能吃掉相当一部分”格式正确但包了壳”的失败案例，剩下真正需要重试的都是内容层面的问题，重试次数和成本都能降下来。

成本上也提一句：max_retries=3 意味着最坏情况下一次业务请求要打 3 次模型调用，按 token 计费的模型这就是 3 倍成本。如果你的 schema 校验失败率长期高于 10%，别一味加重试次数，先回头看 prompt 里的字段说明是不是写得不够清楚——通常是某个字段的取值范围或格式没有用例子讲清楚，模型在”猜”，加一两个示例往往比加重试次数更省钱。

高并发场景下还要注意一个坑：asyncio.sleep(0.5 * (attempt + 1)) 只是让当前协程本身等待，如果你用 asyncio.gather 同时发一批请求，每个请求各自独立重试，不会互相干扰节奏，这没问题；但如果背后调用的是同一个账号的 API Key 且触发了限流，多个协程会同时撞上 429，这时候单个请求的退避策略解决不了根本问题，得在请求发起层加一个全局的并发信号量（比如 asyncio.Semaphore(5)），把整体并发压到限流阈值以下，而不是指望每个请求各自退避就能扛住限流。

流式输出和结构化校验，能不能兼得

这是个绕不开的问题：前端体验要”打字机效果”实时出字，但结构化校验又需要拿到完整 JSON 才能解析。这两者天然有点冲突，处理方式一般分两种。

第一种，也是最省心的做法：结构化输出场景干脆不流式，等模型完整生成完再一次性校验解析。理由很实际——一个半截的 JSON（比如只输出到 {"summary": "这段代码整体）不管拿什么 parser 都解析不出来，你没法对着一个语法都不完整的字符串做 schema 校验。如果你的场景是”生成报告""生成分析结果”这种本来就需要用户等待几秒钟看完整结果的，没必要硬上流式，等待几秒钟展示一个 loading 动画，用户体验的损失远小于半成品 JSON 解析出错的风险。

第二种，如果产品上确实要一边流式显示一边逐步呈现结构化内容（比如实时展示”已识别出第几条 issue”），得用支持增量解析的库，比如 Python 生态里的 partial-json-parser 或者手写一个”容错解析器”：对流式返回的不完整 JSON 字符串，自动补全缺失的闭合括号/引号后再尝试解析，解析失败就跳过这一帧，等下一帧再试。这种方案复杂度和维护成本都高出一截，只有产品体验上确实需要”实时看到结构化内容逐步显现”（比如可视化大屏、长报告实时渲染）才值得投入，一般后台批处理、纯 API 集成场景完全没必要折腾这个。

判断标准很简单：如果你的下游是”程序读取结果做后续处理”（比如写数据库、触发下一步工作流），无脑用非流式；如果下游是”人盯着屏幕看内容逐步生成”，才需要认真评估流式 + 增量解析的投入产出比。

常见问题

schema 里有 required 字段但模型偶尔忽略，怎么处理？ 在 prompt 里把必填字段再次显式声明：“以下字段必须出现：name, score, pass_review”。同时在 schema 中确保这些字段没有默认值，Pydantic 会在缺少时报 ValidationError 并触发重试。实践中发现，这类问题在字段名起得比较抽象时更容易出现——比如叫 flag 而不是 pass_review，模型对字段的用途理解不到位就容易漏填或者填错类型，字段命名越”说人话”，模型遵守的概率越高，这比堆砌重试次数更治本。

Pydantic v1 和 v2 的 schema 格式不兼容怎么处理？ 优先升级到 v2（model_json_schema() 方法），若必须兼容 v1，用 schema() 方法但注意 $defs 字段名差异，function calling 的 schema 需手动适配。另外提一句版本迁移的实际经验：v1 到 v2 的校验器写法也变了（@validator 换成 @field_validator 并且要加 @classmethod），如果你的项目历史包袱重、一次性全切太冒险，可以先用 pydantic.v1 这个兼容命名空间做过渡，把新模块用 v2 写，旧模块暂时保留 v1 写法并存，逐步替换，别指望一次 PR 全量切完不出岔子。

schema 经常变化，如何管理版本？ 把 Pydantic 模型放在独立的 schemas/ 模块，用版本后缀区分（ReviewResultV2），API 调用时明确指定版本，避免自动引用最新版导致线上 schema 突变。更进一步，建议给每个 schema 版本配一条”迁移函数”，把旧版本的数据结构转换成新版本（比如 migrate_v1_to_v2(data: dict) -> dict），这样存量数据不用因为 schema 升级就集体报废，历史记录还能按新 schema 的字段读出来，只是缺失字段填默认值。这个习惯在 schema 迭代频繁、又有历史数据需要兼容读取的项目里能省下大量”数据修复”的加班时间。

校验通过率上不去，是不是该换更贵的模型？ 先别急着换模型，大部分校验失败的根因是 prompt 里的 schema 说明写得含糊，或者 few-shot 示例给得不够。实际经验是，加一到两个”正确输出示例”放进 system prompt，比单纯堆砌 schema 描述文字效果好得多——模型对着示例学格式，比对着抽象的字段说明”猜”格式要稳得多。如果加了示例、优化了字段命名之后校验失败率还是长期高于 5%，再考虑换更强的模型，这时候换模型才是真的对症下药，而不是一遇到失败就砸钱升级。

延伸阅读：大模型应用开发模式 · 应用模式 Hub · 让模型稳定输出 JSON · RAG 切块策略