结构化输出

概述

JSON Schema 约束与输出解析的容错机制

结构化输出的本质是借助 LLM 的 Function Calling 能力（或 JSON Mode）强制模型生成符合特定 Schema 的数据。Agents SDK 通过 output_type 参数接受 Pydantic 模型，SDK 会自动将其转换为 JSON Schema 并注入到 system prompt 中。然而，即使有了 Schema 约束，模型仍可能生成格式错误的 JSON——尤其是在使用轻量级模型或处理复杂嵌套结构时。

SDK 内部的处理流程如下：

1. Schema 生成：将 output_type 的 Pydantic 模型转换为 JSON Schema，通过 response_format={"type": "json_schema", "schema": ...} 传递给 API。
2. 模型生成：LLM 在生成文本时受到 Schema 的语法约束，理论上不会产出非法 JSON。
3. 解析验证：SDK 使用 Pydantic 的 model_validate_json 对模型输出进行反序列化和校验。
4. 错误处理：如果解析失败，SDK 会将错误信息返回给模型，请求其修正（部分版本支持自动重试）。

从设计思想来看，这种"约束生成 + 严格校验"的双保险模式比单纯的文本解析（如正则提取 JSON）要健壮得多。但与专门的外部库如 instructor 相比，Agents SDK 的重试和修正机制较为简单，不支持自动 few-shot 纠错或递归修正。

在性能层面，JSON Schema 约束会增加输入 token 数量（因为 Schema 本身需要被编码到 prompt 中）。对于复杂模型（数十个字段、多层嵌套），Schema 可能占据数百个 token，这在高频调用场景下不可忽视。

使用 Pydantic 模型定义 output_type，让 Agent 返回类型安全的数据结构。

正文

相关阅读

• OpenAI Agents SDK 快速上手
• SDK 核心概念一览

参考文档

• OpenAI Agents SDK 官方文档
• GitHub 仓库

完整实战示例：健壮的结构化数据提取流水线

以下示例展示了如何在生产环境中处理结构化输出，包括 Schema 校验失败时的降级处理和日志记录：

import json
import asyncio
from typing import Optional
from pydantic import BaseModel, Field, ValidationError
from agents import Agent, Runner, RunContextWrapper


class Address(BaseModel):
    city: str = Field(description="城市名称")
    district: str = Field(description="区县名称")
    street: str = Field(description="街道地址")
    zip_code: Optional[str] = Field(default=None, description="邮政编码")


class UserProfile(BaseModel):
    name: str = Field(min_length=1, max_length=50, description="用户姓名")
    age: int = Field(ge=0, le=150, description="年龄")
    email: str = Field(pattern=r"^[\w\.-]+@[\w\.-]+\.\w+$", description="邮箱地址")
    address: Address = Field(description="居住地址")
    tags: list[str] = Field(default_factory=list, max_length=10, description="兴趣标签")


class ExtractResult(BaseModel):
    success: bool
    data: Optional[UserProfile] = None
    raw_output: Optional[str] = None
    error_message: Optional[str] = None


async def extract_user_profile(text: str, max_retries: int = 2) -> ExtractResult:
    """从自由文本中提取用户画像，支持失败重试和降级。"""
    agent = Agent(
        name="Extractor",
        instructions="""
从用户提供的文本中提取结构化信息。必须严格按照要求的 JSON 格式输出  #不要添加任何额外说明。
如果某个字段无法确定，使用合理的默认值或空字符串，不要编造信息。
""".strip(),
        model="gpt-5-nano",
        output_type=UserProfile)

结构化输出解析流程

下图展示了从 LLM 原始输出到强类型对象的完整解析与验证流程：
mermaid
flowchart TD
A[LLM 原始输出] --> B{格式检查}
B -->|JSON 格式| C[Syntax 解析]
B -->|Markdown 包裹| D[去除标记]
D --> C
B -->|纯文本| E[二次提示修正]
E --> A
C --> F[Pydantic 校验]
F -->|通过| G[返回类型化对象]
F -->|字段缺失| H[默认值填充]
F -->|类型错误| I[自动转换尝试]
H --> G
I -->|成功| G
I -->|失败| J[返回解析错误]
style G fill:#c5e0b4,stroke:#5a4a3a
style J fill:#f4b183,stroke:#5a4a3a

实际生产中，LLM 输出格式的不稳定性是结构化解析的最大挑战。即使明确指示请以 JSON 输出，模型仍有约 5-10% 的概率在 JSON 前后添加解释性文字或 Markdown 标记。

## 结构化输出的设计模式

设计良好的输出 schema 能显著提升解析成功率和后续处理效率。以下是四个关键设计原则：

**原则一：字段命名语义化**。使用 delivery_address 而非 field_1，让 LLM 能根据字段名推断期望的内容类型。同时，为每个字段添加 description 参数，进一步消除歧义。

```python
from pydantic import BaseModel, Field

class OrderInfo(BaseModel):
    product_name: str = Field(description="商品全称，如 iPhone 16 Pro")
    quantity: int = Field(description="购买数量，正整数")
    price: float = Field(description="单价，保留两位小数")

原则二：嵌套层级扁平化。尽量避免超过 3 层的嵌套结构。深层嵌套不仅增加 LLM 出错概率，也使得后续代码处理变得复杂。如果业务逻辑确实需要深层结构，考虑拆分为多个独立的 schema，分多次调用获取。

原则三：枚举值优于自由文本。对于有限取值的字段（如状态、类别），使用 Literal 或 Enum 定义允许的取值范围。这能将模型的输出空间从无限缩小到有限，大幅降低幻觉概率。

原则四：容错设计。为每个字段设置合理的默认值，并标记 Optional 字段。当 LLM 遗漏某个字段时，程序可以使用默认值继续运行，而非直接抛出异常中断流程。对于关键字段，可以在 schema 校验失败后触发重试机制，给 LLM 一次修正的机会。

实际生产中，LLM 输出格式的不稳定性是结构化解析的最大挑战。即使明确指示请以 JSON 输出，模型仍有约 5-10% 的概率在 JSON 前后添加解释性文字或 Markdown 标记。健壮的解析器需要处理这些边缘情况：清理可能的 Markdown 标记、捕获 ValidationError 后自动重试。对于特别关键的字段，可以在 schema 中设置 Field(description=…) 提供详细说明，同时在指令中强调输出格式要求，双重保障降低解析失败率。

常见问题与调试

问题一：模型输出多余的解释文本

即使设置了 output_type，某些模型仍可能在 JSON 前后添加解释性文字（如 "Here is the result:"），导致 Pydantic 解析失败。解决方案：

1. 在 instructions 中明确禁止额外文本："Output ONLY valid JSON. No markdown, no explanations."
2. 使用 response_format={"type": "json_object"}（OpenAI 原生支持）来强制纯 JSON 输出。
3. 在后处理中使用正则提取第一个 { 和最后一个 } 之间的内容作为容错解析。

问题二：枚举值或 Literal 类型不生效

Pydantic 的 Literal 类型在转换为 JSON Schema 后会生成 enum 约束，但轻量级模型对 enum 的遵守率较低。建议：

1. 在 instructions 中重复枚举选项的文字描述。
2. 在后置 Guardrails 中校验枚举值，如果不匹配则映射到默认值。
3. 考虑使用更大的模型来处理严格的枚举约束。

问题三：嵌套模型导致 Schema 过大

深层嵌套的 Pydantic 模型会生成庞大的 JSON Schema，可能超出 API 的 Schema 大小限制（通常为 16KB）。优化方法：

1. 扁平化数据结构，减少嵌套层级。
2. 将大模型拆分为多个小模型，分多次提取。
3. 使用 $ref 和 definitions 复用子结构，减少重复定义。

与其他方案对比

维度	Agents SDK output_type	Instructor 库	LangChain Output Parser
校验时机	生成时约束 + 解析时校验	生成时约束 + 自动重试修正	生成后文本解析
重试机制	基础（需手动实现）	高级（自动反馈错误）	需手动实现
Schema 定义	Pydantic 原生	Pydantic 原生	Pydantic 或自定义
适用场景	标准结构化提取	高可靠要求的复杂提取	简单字段提取

Instructor 是目前结构化输出领域最成熟的库，它通过"重试-修正"循环显著提升了复杂 Schema 的提取成功率。Agents SDK 的内置 output_type 对于中等复杂度的场景已经足够，但在金融、医疗等对准确性要求极高的领域，建议结合 Instructor 使用，或在应用层自行实现重试逻辑。

结构化输出解析流水线

---
title: 结构化输出校验与容错重试流程
---
flowchart TD
    A[用户输入文本] --> B[Agent调用]
    B --> C[LLM生成JSON]
    C --> D[Pydantic Schema校验]
    D -->|校验通过| E[返回结构化对象]
    D -->|校验失败| F[错误分类]
    F -->|格式错误| G[正则提取JSON片段]
    F -->|字段缺失| H[填充默认值]
    F -->|类型不匹配| I[类型强制转换]
    G --> J{修复后重试}
    H --> J
    I --> J
    J -->|成功| E
    J -->|失败| K[记录错误日志]
    K --> L[返回降级结果]

Schema版本演进与多模型兼容性处理

结构化输出的Schema在实际业务中不可避免地会经历迭代。新增字段、修改约束条件、调整嵌套结构都是常见需求。如果前后端Schema不一致，可能导致解析失败或数据丢失。因此，建立Schema版本管理机制至关重要。版本管理的核心思想是向后兼容：小版本号增加表示新增可选字段，大版本号增加表示破坏性变更，消费者必须显式升级。同时，Schema的体积直接影响API调用成本，深层嵌套模型可能生成超过16KB的JSON Schema，超出部分会被截断或拒绝。

另一个容易被忽视的问题是不同模型对JSON Schema的支持程度不同。轻量级模型可能在处理复杂的anyOf、oneOf或深层嵌套时表现不佳。以下是一个支持版本演进和多模型适配的Schema管理方案：

from typing import Type, Optional
from pydantic import BaseModel, Field
import json


class SchemaVersion:
    """Schema版本标识符，遵循SemVer规范。"""
    def __init__(self, major: int, minor: int, patch: int = 0):
        self.major = major
        self.minor = minor
        self.patch = patch
    
    def __str__(self):
        return f"{self.major}.{self.minor}.{self.patch}"
    
    def is_compatible_with(self, other: "SchemaVersion") -> bool:
        return self.major == other.major


class VersionedSchemaManager:
    """管理多版本Pydantic Schema，支持根据模型能力自动降级。"""
    
    def __init__(self):
        self._schemas: dict[str, Type[BaseModel]] = {}
        self._versions: dict[str, SchemaVersion] = {}
    
    def register(self, name: str, version: str, schema_cls: Type[BaseModel]):
        self._schemas[f"{name}@{version}"] = schema_cls
        v_parts = version.split(".")
        self._versions[name] = SchemaVersion(
            int(v_parts[0]), int(v_parts[1]),
            int(v_parts[2]) if len(v_parts) > 2 else 0
        )
    
    def get_schema(self, name: str, version: Optional[str] = None,
                   model_capability: str = "standard") -> Type[BaseModel]:
        key = f"{name}@{version}" if version else None
        
        if key and key in self._schemas:
            schema = self._schemas[key]
        else:
            candidates = [k for k in self._schemas
                          if k.startswith(f"{name}@")]
            if not candidates:
                raise KeyError(f"No schema registered for {name}")
            key = sorted(candidates)[-1]
            schema = self._schemas[key]
        
        if model_capability == "lite":
            return self._simplify_schema(schema)
        return schema
    
    def _simplify_schema(self, schema_cls: Type[BaseModel]) -> Type[BaseModel]:
        return schema_cls
    
    def generate_json_schema(self, name: str, version: Optional[str] = None,
                             model_capability: str = "standard") -> dict:
        schema_cls = self.get_schema(name, version, model_capability)
        return schema_cls.model_json_schema()

结构化输出错误的自动修复策略

即使有了Schema约束，模型仍可能生成不符合要求的输出。生产级系统不能简单地将解析错误抛给用户，而应具备自动修复能力。常见的修复策略包括：正则提取被Markdown包裹的JSON、用默认值填充缺失字段、以及对类型错误进行强制转换。修复失败时才降级到原始文本输出。修复策略的排序很重要：应先尝试格式提取（最轻量），再尝试字段填充，最后才考虑类型强制转换（风险最高）。过度修复可能掩盖模型质量问题，因此所有修复操作都必须记录到日志中供后续分析。

以下是一个多阶段错误恢复处理器：

import re
import json
from pydantic import BaseModel, ValidationError
from typing import Type


class OutputRepairPipeline:
    """多阶段结构化输出修复流水线。"""
    
    def __init__(self, schema_cls: Type[BaseModel]):
        self.schema_cls = schema_cls
    
    def _extract_json(self, text: str) -> str:
        patterns = [
            r"```json\s*(.*?)\s*```",
            r"```\s*(.*?)\s*```",
            r"(\{.*\})",
        ]
        for pat in patterns:
            match = re.search(pat, text, re.DOTALL)
            if match:
                return match.group(1).strip()
        return text.strip()
    
    def _coerce_types(self, data: dict) -> dict:
        fields = self.schema_cls.model_fields
        for key, field_info in fields.items():
            if key not in data:
                continue
            expected_type = field_info.annotation
            value = data[key]
            if expected_type is int and isinstance(value, str):
                try:
                    data[key] = int(value)
                except ValueError:
                    data[key] = field_info.default or 0
        return data
    
    def repair(self, raw_output: str) -> BaseModel:
        extracted = self._extract_json(raw_output)
        try:
            data = json.loads(extracted)
        except json.JSONDecodeError:
            raise ValueError("无法提取有效JSON")
        
        data = self._coerce_types(data)
        try:
            return self.schema_cls.model_validate(data)
        except ValidationError as e:
            defaults = {k: v.default
                        for k, v in self.schema_cls.model_fields.items()
                        if v.default is not None}
            for key in self.schema_cls.model_fields:
                if key not in data:
                    data[key] = defaults.get(key)
            return self.schema_cls.model_validate(data)

要点总结：

1. Schema版本号遵循SemVer，主版本变更表示不兼容的破坏性修改；兼容性检查应在服务启动时完成，避免运行时才发现不匹配。
2. 为轻量模型提供简化版Schema，可显著提高结构化输出的解析成功率；建议移除pattern和深层嵌套约束，使用更宽松的字段定义。
3. 修复流水线应按从轻到重的顺序执行：提取JSON片段、类型强制转换、默认值填充；每步失败才进入下一步，避免过度修复引入错误数据。
4. 建议在CI流水线中增加Schema变更检测，自动生成迁移文档和兼容性报告；所有修复操作应记录到日志以便后续分析模型质量。
5. 对于金融、医疗等高可靠性领域，建议结合双重校验策略：两个不同模型分别提取，结果不一致时转人工审核。

生产环境部署与性能优化

Schema 治理的实践要点

将本章节的技术应用到生产环境时，首要考虑的是稳定性与可观测性。建议采用渐进式 rollout 策略：先在开发环境验证核心逻辑，再迁移到预发布环境进行压力测试，最后才全量上线。部署过程中应配置完善的日志收集和指标监控，确保任何问题都能被快速发现和定位。

具体来说，需要在基础设施层面做好以下准备：容器资源限制（CPU/内存）、网络策略配置（防火墙规则、服务网格）、持久化存储选型（SSD vs 标准盘）以及备份恢复方案。对于高可用要求严格的场景，建议部署多实例并配置负载均衡，避免单点故障导致服务中断。

输出准确率追踪的关键指标

监控是生产系统的生命线。针对本章节涉及的功能，建议重点跟踪以下指标：请求延迟（P50/P95/P99）、错误率（4xx/5xx/超时）、吞吐量（QPS/TPS）以及资源利用率（CPU/内存/磁盘/网络）。这些指标应接入统一的监控大盘，并设置合理的告警阈值。

除了基础指标，还应关注业务层面的指标。例如功能成功率、用户满意度、成本消耗趋势等。通过将技术指标与业务指标关联分析，可以更准确地评估系统改进的实际价值，避免陷入"为了优化而优化"的陷阱。

批量数据提取的架构考量

随着业务规模增长，单实例部署很快会成为瓶颈。扩展性设计应在项目初期就纳入考量，而非事后补救。水平扩展通常比垂直扩展更具成本效益，但也引入了分布式系统的复杂性（数据一致性、服务发现、负载均衡等）。

在扩展过程中，建议遵循"无状态优先"原则：将状态外置到独立的存储层（如 Redis、PostgreSQL），使计算层可以随时水平扩容。对于无法避免的状态（如会话、缓存），采用分布式一致性协议或最终一致性模型来管理。定期进行容量规划和压力测试，确保系统在流量峰值时仍能稳定运行。

运维团队的协作建议

技术方案的落地离不开高效的团队协作。建议建立清晰的运维手册（Runbook），涵盖常见故障的诊断步骤、应急处理流程和升级路径。同时，通过定期的复盘会议，将线上事故转化为团队的学习素材，持续完善系统的健壮性。

在工具链方面，推荐将本章节的配置和脚本纳入版本控制（Git），并使用 Infrastructure as Code（IaC）工具（如 Terraform、Ansible）管理基础设施变更。这不仅能提高部署效率，还能确保环境一致性，减少"在我机器上能跑"的问题。

结构化输出的失败率往往被低估。在上线前，建议收集 1000 条真实用户输入进行压力测试，统计 JSON 解析失败率。如果超过 1%，就需要加强 Schema 约束或更换更强模型。

对于需要高度可靠性的场景，可以考虑使用双重校验策略：让两个不同的模型分别提取结构化数据，然后对比结果。当两者不一致时，标记为低置信度并转人工审核。

在实际项目中，结构化输出的 Schema 往往会随着业务需求演化。建议在 Schema 定义中预留扩展字段（如 metadata、extra_info），以便未来新增字段时无需修改现有解析逻辑，保持向后兼容。