第 1 章讲义 · 04 Structured Output 结构化输出

配套代码：结构化引擎 src/anna/structured.py、网关端点 /v1/structured （src/anna/server/app.py）、三家 provider 的原生结构化输出实现、 首页"结构化输出"面板

学完本节，你应该能回答：为什么 Agent 不能靠自然语言解析？JSON Schema 怎么 定义？Pydantic 和 Schema 是什么关系？模型原生结构化输出怎么用？校验失败 怎么纠错、怎么降级？

目录#

1. 为什么不能依赖自然语言解析
2. JSON Schema：定义输出结构
3. Pydantic Model：定义 + 校验 + 反序列化
4. 模型原生结构化输出
5. Schema 校验
6. 输出纠错与重试
7. 结构化输出失败处理：降级与回退
8. 输出格式与业务协议设计

1. 为什么不能依赖自然语言解析#

Agent 系统里，模型输出要喂给代码：查词结果要填进卡片、工具参数要发请求、 SQL 要执行。自然语言解析的典型翻车：

模型回复："好的，单词是 hello，意思是 你好，例句是 Hello world！"

正则/切片解析这种输出，任何换行、加粗、多余解释都会破坏结果；更糟的是 解析失败时你不知道它是格式错了还是答案错了。所以工程原则是：

• 输出协议先行：先定义"模型必须返回什么形状"，再让它生成；
• 校验兜底：返回后必须校验，而不是信任；
• 失败有出路：校验不过就纠错重试或降级，绝不把脏数据传给下游。

2. JSON Schema：定义输出结构#

JSON Schema 是描述 JSON 结构的标准：字段名、类型、是否必填、取值范围。它是 跨语言的协议语言——模型、Python、前端都能理解。

{
  "type": "object",
  "required": ["word", "meaning", "example"],
  "properties": {
    "word": {"type": "string"},
    "meaning": {"type": "string"},
    "phonetic": {"type": "string"},
    "example": {"type": "string"}
  },
  "additionalProperties": false
}

要点：

• type：对象/数组/字符串/数字/布尔；
• required：必填字段，缺失即校验失败；
• additionalProperties: false：禁止多余字段（防止模型自作主张加东西）；
• 数组可加 minItems/maxItems/items，字符串可加 enum 限定取值。

本仓库的 /v1/structured 直接接收 JSON Schema：

result = structured_completion(
    provider,
    messages,
    json_schema=WORD_CARD_SCHEMA,   # 上面的 JSON
    max_attempts=3,
    fallback={"word": "", "meaning": "查询失败", "example": ""},
)

3. Pydantic Model：定义 + 校验 + 反序列化#

Pydantic 是 Python 侧的 Schema：用类定义结构，model_validate 校验并转成 Python 对象。JSON Schema 描述"协议长什么样"，Pydantic 让"代码直接拿到对象"。

from pydantic import BaseModel
class WordCard(BaseModel):
    word: str
    meaning: str
    phonetic: str = ""
    example: str = ""
card = WordCard.model_validate({"word": "hello", "meaning": "你好"})
print(card.word, card.meaning)  # hello 你好
try:
    WordCard.model_validate({"word": "hello"})  # 缺 meaning
except Exception as exc:
    print(exc)  # ValidationError: meaning Field required

本仓库中 Pydantic 用于网关请求入口（server/schemas.py 的 ChatRequest/StructuredRequest，FastAPI 自动校验），响应出口的校验由 structured.py 的 Schema 校验完成。代码里要接 Pydantic 时，把模型类换成 上面的 WordCard 即可——引擎内部用的是协议无关的 JSON Schema。

4. 模型原生结构化输出#

在 API 层直接把 Schema 传给模型，让模型侧保证输出合规，而不是靠 prompt 求它：

raw = provider.complete(
    messages,
    response_format={"type": "json_schema", "json_schema": WORD_CARD_SCHEMA},
)

三家厂商的原生能力不同，本仓库的 adapter 已屏蔽差异：

对应测试在 tests/test_structured_output.py。关键认知：原生结构化输出把 “输出合规"从 prompt 软约束升级成 API 硬约束，但仍要校验兜底——模型可能 超长截断、返回空、或违反 schema。

5. Schema 校验#

模型返回后必须校验。structured.py：

def validate_schema(data, schema) -> None:
    jsonschema.validate(instance=data, schema=schema)

配合宽容的 JSON 提取（容忍 ```json 围栏和前后说明文字）：

def extract_json(text):
    """去掉围栏；用平衡括号扫描找到第一个完整 JSON 对象/数组。"""
    ...

校验失败会抛 jsonschema.ValidationError（缺字段、类型错、多余字段）， 错误信息正好用作第 6 节的"反喂"素材。

6. 输出纠错与重试#

校验失败时，把模型自己的输出 + 校验错误反喂给它，让它自行修正——这比 无脑重发效果好得多，因为模型能看到自己哪里错了：

working_messages = [
    *working_messages,
    {"role": "assistant", "content": raw},      # 模型自己的输出
    {"role": "user", "content":                 # 校验错误 + 修正指令
        f"上面的输出不符合要求：{last_error}\n"
        "请重新输出，只输出符合 JSON Schema 的 JSON，不要任何解释。"},
]

structured_completion 的重试循环：

for attempt in range(max_attempts):
    raw = provider.complete(working_messages, response_format={"type": "json_schema", "json_schema": json_schema})
    try:
        data = extract_json(raw)
        validate_schema(data, json_schema)
        return StructuredResult(data=data, raw=raw, attempts=attempt + 1)
    except (...) as exc:
        last_error = str(exc)
        working_messages = [...反喂...]

测试 tests/test_structured.py::StructuredEngineTest.test_retry_feeds_error_back 验证了第二次调用时 messages 里确实带了"模型输出 + 错误反馈”。

7. 结构化输出失败处理：降级与回退#

重试次数耗尽后有两种出路：

1. 降级到缓存/默认值（推荐）：调用方传入 fallback，返回 degraded=True，让上游知道这是降级数据而不是模型结果；
2. 直接失败：抛 StructuredOutputError，网关返回 422。

result = structured_completion(
    provider, messages, json_schema,
    max_attempts=3,
    fallback={"word": "cache", "meaning": "查询失败，返回缓存", "example": ""},
)
if result.degraded:
    log.warning("结构化输出降级：%s", result.error)

首页"结构化输出"面板有"故意失败"预设（minItems: 999999 不可满足的 Schema）， 可以直观看到：3 次尝试全部失败 → 返回 fallback + degraded 标记。

8. 输出格式与业务协议设计#

输出格式就是你和模型之间的业务协议，设计建议：

• Schema 要版本化：字段演进不要原地改，加 $schema 或版本号，避免下游 悄悄破坏；
• 枚举收窄取值：状态类字段用 enum（pending/done/error），不要自由 文本；
• 容忍缺失：必填字段越少越稳；可选字段给默认值（Pydantic 的 = ""）；
• additionalProperties 按需：协议内关掉（防乱加），需要扩展性时保留并 靠校验兜底；
• 降级可见：所有"用了缓存/默认值"的响应都要带标记（degraded）， 这是可观测性的一部分。

小结#

• Agent 系统必须"协议先行 + 校验兜底"，不能靠自然语言解析；
• JSON Schema 是跨语言协议，Pydantic 是 Python 侧定义 + 校验 + 反序列化；
• 原生结构化输出把约束从 prompt 软约束升级为 API 硬约束，但仍有边界；
• 返回后必须校验；失败把"输出 + 错误"反喂给模型重试；
• 重试耗尽 → 降级缓存/默认值（带 degraded 标记）或明确失败；
• 输出格式是业务协议：版本化、收窄枚举、容忍缺失、降级可见。

动手练习：

1. 首页"结构化输出"选"单词卡"预设跑一次，观察 attempts 和校验结果；
2. 选"故意失败"预设，看 3 次尝试后如何降级并返回 fallback；
3. 用 /v1/structured 传一个必填 age: integer 的 schema，观察模型输出 与校验结果；
4. 把 max_attempts 调成 1，对比失败行为（无 fallback 时返回 422）。

附录：讲义内容 × 项目实现对照#

图例：✅ 已实现并有测试 / ⬜ 未实现（讲义中已注明或属后续阶段）/ 📘 通用知识或示例说明。