防弹结构体：Pydantic、Zod 与大模型 Native Structured Output 的底层较量

(第 49 篇：Agent 协议之盾)

我们在前几篇讨论了如果抓取了错误 JSON 应该怎么“惩罚” Agent。但这其实属于“后置补救”。在当代最新的 Agent 架构中，解决这一痛点的方案已经发生了翻天覆地的跃进：从“期望模型生成好再解析”，演进到了“在生成级别 (Generation Level) 就拿枪指着模型，强制走入防弹结构体”。

本篇将深入探讨如何利用强类型 Schema 构筑 Agent 的逻辑边界，并解析 Native Structured Output 是如何从物理层面消除 JSON 异常的。

0. 校验不是一层，是两层：生成期约束 + 执行期闸门

很多系统只做了一件事：让模型输出“看起来像 JSON”。 这会让你在生产上迟早踩到两个坑：

1. JSON 结构正确，但语义危险（比如路径越权、命令注入）。
2. schema 严格，但遇到拒答/截断后你的程序没有失败路径。

正确做法是两层校验：

本文讨论的 schema 只是协议边界，不是安全边界。

1. 约束演进：从“祈祷”到“铁律”

在 Agent 协议的发展史上，我们经历了四个阶段：

1. 祈祷阶段：在 Prompt 里加一句 "Please return JSON"。失败率：30%+。
2. 正则阶段：用正则表达式暴力提取 Markdown 中的代码块。失败率：15%。
3. 后置校验阶段：引入 Pydantic/Zod 解析，失败则扔回报错让模型重试。失败率：5%（消耗大量 Token）。
4. 受限解码阶段 (Constrained Decoding)：通过底层算力控制 Token 出口，失败率：0%。

2. Pydantic / Zod：Single Source of Truth

在工业级 Agent 开发中，Schema 首先是强类型代码，其次才会被翻译给 LLM。这种 单一事实来源 确保了你的文档和代码逻辑永远同步。

2.1 【代码实战】构建一个防弹的 Shell 执行 Schema

让我们在 Python 中使用 Pydantic 定义一个安全的 Shell 工具：

from pydantic import BaseModel, Field, validator
import re

class ShellAction(BaseModel):
    """
    定义一个安全的 Shell 执行契约。
    模型必须严格遵循此结构，否则在生成阶段就会被阻断。
    """
    command: str = Field(..., description="要执行的 shell 命令")
    working_dir: str = Field(default="/tmp/workspace", description="执行目录，必须是绝对路径")
    timeout_seconds: int = Field(default=30, ge=1, le=300)

    @validator("command")
    def prevent_suicidal_commands(cls, v):
        # 即使模型想作恶，这一层代码校验是最后的物理关卡
        forbidden = ["rm -rf /", "mkfs", "dd"]
        if any(f in v for f in forbidden):
            raise ValueError("检测到破坏性系统命令，执行已被底层内核阻断！")
        return v

当你把这个 Pydantic 模型作为工具投喂给大模型时，它会转化为一套 JSON Schema。大模型看到的不仅仅是“命令”两个字，它看到的是包含类型、范围、正则约束的严密法律。

2.2 首请求延迟与 schema 缓存：你需要观测它

严格 schema 往往需要在第一次使用时做额外处理（例如缓存/编译产物），这会带来首请求延迟。 如果你不观测它，你上线后会把“慢”误诊成“模型变笨”。

建议至少记录：

这些字段进入 trace/span 与审计，才能支撑长期稳定运行。

3. 核心内幕：Native Structured Output 的物理原理

当我们在调用 GPT-4o 或 Gemini 1.5 的 response_format: { "type": "json_schema" } 时，API 的解码环节发生了什么？

3.1 受限解码 (Constrained Decoding) 技术

这就是所谓的 Logit Bias / Grammar Guiding。

1. 概率矩阵预测：当模型生成到 {"age": 这 7 个字符后。
2. 动态掩码 (Masking)：大模型原本可能预测下一个 Token 是 "twenty", (空格), 25, 或 [ (中括号)。
3. 物理干预：因为你的 Schema 规定了 age 必须是 integer。API 底层的解码器会瞬间将所有非数字符号的 Token 概率强行修正为 0。
4. 必中结果：模型在物理层面上“吐”出的下一个 Token 只能且必须是数字。

结论：在 Native 模式下，你再也不需要担心 JSON 多了一个逗号或少了一个括号。因为非法的中括号在那个步进点上根本无法通过预测矩阵。

4. 后置反馈回路：针对非 Native 模型的兜底

如果你在使用本地 Llama 3 或 Qwen，且没有使用 llama.cpp 的 Grammar 控制功能，那么你必须实现一套高度健壮的 Validation-Feedback 管道。

async def orchestrate_agent(task):
    context = [{"role": "system", "content": "You are a JSON machine."}]
    
    for _ in range(3): # 最高 3 次重试机会
        response = await llm.chat(context)
        try:
            # 尝试 Pydantic 强转
            action = ShellAction.parse_raw(response.content)
            return await execute_physical(action)
        except ValidationError as e:
            # 极其关键：将结构化的 Pydantic 报错信息格式化
            error_feedback = f"""
            [VALIDATION_ERROR] 你生成的 JSON 不符合 Schema 契约：
            {e.json()}
            请检查：
            - timeout_seconds 必须是整数。
            - 不能包含禁用命令。
            请再次尝试修改你的输出。
            """
            context.append({"role": "user", "content": error_feedback})
            continue

极客提示：不要只是把 Exception 扔过去。要把 Pydantic 的 e.json() 扔过去，因为它包含了精确的键路径（Keypath）和错误原因，大模型的纠错效率会提升一个量级。

6. 失败路径：拒答、截断、部分输出

严格结构化输出也会失败，常见边界条件是：

• 拒答：模型因为安全策略拒绝回答。
• 截断：达到 max_tokens 导致 JSON 未闭合。
• 部分输出：只输出了部分字段。

工程上必须明确如何处理：

1. 进入重试前，先判断是否会产生副作用（幂等 key 必须绑定）。
2. 重试必须有最大次数与退避，避免重试风暴。
3. 失败必须写入审计与 trace/span，才能复盘为什么进入降级。

7. 断路器与幂等：校验失败不等于可以无限重试

校验失败是常态，不是异常。 但“无限重试”是事故制造机，尤其在工具会产生副作用时。

最小治理建议：

• 断路器：同一类校验错误连续 3 次，停止写入型工具，进入只读诊断（shadow mode）。
• 幂等：任何副作用工具调用都必须绑定 idempotency_key，否则重试就是重复提交。
• 审计：记录 schema_id、错误类别、重试次数、最终降级策略，支持复盘。

8. 最小验收清单：你怎么证明“结构化输出”真的稳定

把下面清单写进你的 CI 或回归测试里，别只靠肉眼：

5. 极端挑战：处理流式中的“脏”数据

当 Agent 在进行大批量文件读写时，JSON 可能会非常巨大。如果你想在 token 还没跳完时就进行拦截（PII 过滤、非法路径监测），你需要一套 Streaming JSON Parser。

它维护一个 Stack（栈）架构：

• 遇到 { 压栈。
• 遇到 } 出栈。
• 实时监控当前正在处于哪个 key 之下。 一旦当前 Key 是 path，且其 Value 字符串开始匹配敏感目录，拦截器可以立即切断 TCP 连接，实现物理级的实时防爆。

本章精粹

1. Schema 是 Agent 的“三观”：没有 Schema 约束的 Agent 就像是没有灵魂的脱缰野马。
2. Native 是首选：如果你购买的是商业 API，务必开启 strict: true，这能省下你 20% 的重试 Token 成本。
3. 精准反馈：错误反馈不仅是字符串，更是带路径的“逻辑补丁”。

学会了用结构化输出武装你的 Agent，它就从一个“会偶尔精神错乱的小助手”，变成了一台毫无感情、永不越界的高精密数控机床。

下一章，我们将讨论一个极其棘手的实际问题：【从裸文本中暴力提取指令：当模型拒绝遵循 JSON 格式时，我们该如何使用正则和宽容解析技术实现软拦截？】。我们要去写那些“不那么优雅但能救命”的代码了。

(本文完 - 深度解析系列 15 / 全文约 1600 字) (注：建议在 IDE 中运行 pydantic.BaseModel.schema_json()，观察代码是如何转化为机器可读契约的。)