ELL3N HSU

Available in Chinese only.

← 文章

JSON 合法以后:结构化输出与工具执行契约

以日志查询和创建外部对象为例,区分 Schema、业务校验、权限和副作用状态,说明函数调用为什么不能替运行时完成验收。

一个日志查询工具收到 start、end、service 和 limit 四个字段。JSON 可以解析,字段类型也符合 Schema,仍可能出现起止时间颠倒、查询跨度过大,或请求访问当前账号无权查看的服务。

假设 start 比 end 晚,模型给出的仍然可以是一份格式正确的 JSON,但后端无法按它执行查询。要把这类调用接到真实服务,需要先检查结构,再检查时间范围和权限。下面先把日志查询从原始响应跑到可执行参数,再看看换成写操作后,超时和重试会多出什么问题。

三个容易混在一起的接口层次

机制主要职责不自动保证什么
结构化输出让输出符合约定格式或 Schema内容真实、业务合理、操作获准
函数或工具调用模型提出工具名称与参数,交由程序处理后端已执行或一定成功
MCP客户端与工具服务之间的能力发现、调用等协议应用已选好工具,或当前用户有执行权限

不同模型 API 支持的 Schema 子集和约束方式不同,不能把“支持 JSON”当作相同承诺。即使平台提供严格结构约束,运行时仍应处理拒绝、截断、协议错误与版本不匹配。

工具数量多时,先解决模型如何发现定义。这篇从定义已经送到模型、调用已经返回的地方继续。

Schema 把接口形状写清楚

先把这四个字段写成 Schema。这样可以在进入查询逻辑前,检查模型是否漏了服务名、拼错了字段,或给了错误的参数类型:

{
  "type": "object",
  "properties": {
    "service": { "type": "string", "minLength": 1 },
    "start": { "type": "string" },
    "end": { "type": "string" },
    "limit": { "type": "integer", "minimum": 1, "maximum": 200 }
  },
  "required": ["service", "start", "end", "limit"],
  "additionalProperties": false
}

properties 描述字段,并不单独要求字段出现;required 承担必填约束。additionalProperties: false 拒绝未声明字段,避免拼错参数名后被静默忽略。这里故意把时间写成字符串,日期语法与时区由应用显式解析,不依赖不同校验器对格式关键字的默认行为。

字段合法以后,仍需要比较两个时间、检查跨度,并从可信身份中取得可访问服务列表。这个权限列表不能由模型在参数里自行声明。

先从原始响应建立校验入口

上面的 Schema 是接口文档的一部分。运行时需要真正执行校验,不能只把 Schema 放进 Prompt,然后假定模型会遵守。下面用标准库实现本例的四字段检查,不实现通用 JSON Schema 解释器;已有 Schema 校验库时可以用对应校验器替换这一层。

import json

def unique_object(pairs):
    result = {}
    for key, value in pairs:
        if key in result:
            raise ValueError("duplicate field: " + key)
        result[key] = value
    return result


def reject_constant(value):
    raise ValueError("non-JSON constant: " + value)


def decode_query(raw):
    args = json.loads(raw, object_pairs_hook=unique_object,
                      parse_constant=reject_constant)
    required = {"service", "start", "end", "limit"}
    if type(args) is not dict or set(args) != required:
        raise ValueError("expected exactly service, start, end, limit")
    for field in ("service", "start", "end"):
        if type(args[field]) is not str:
            raise ValueError(field + " must be a string")
    if not args["service"]:
        raise ValueError("service must not be empty")
    if type(args["limit"]) is not int or not 1 <= args["limit"] <= 200:
        raise ValueError("limit must be an integer from 1 to 200")
    return args

这里有两个容易漏掉的输入。{"limit": true} 在 Python 中满足 isinstance(True, int),但不是本接口要接受的整数数量,所以使用精确类型检查。这也使本例比 JSON Schema 的整数约束更严格:20.0 在 JSON Schema 中可作为整数,本地入口只接受解码为 int 的写法。重复字段要在构造字典前识别;解析成字典之后,前一个同名字段就已经丢失。示例另外拒绝 Python JSON 解码器默认允许的 NaN 等常量。服务入口还应限制请求体大小。

把业务约束放在可测试的函数里

将 decode_query 的结果交给下面的函数,继续检查业务约束。示例接口约定只接收形如 2026-10-13T08:30:00Z 的 UTC 秒级时间,不接受本地时间或其他格式;最大跨度由后端配置传入。

from datetime import datetime, timedelta, timezone
import re


def parse_utc(value):
    pattern = r"[0-9]{4}-[0-9]{2}-[0-9]{2}T[0-9]{2}:[0-9]{2}:[0-9]{2}Z"
    if re.fullmatch(pattern, value) is None:
        raise ValueError("时间必须是 UTC 秒级格式")
    return datetime.strptime(value, "%Y-%m-%dT%H:%M:%SZ").replace(
        tzinfo=timezone.utc
    )


def validate_query(args, allowed_services, max_span=timedelta(hours=1)):
    if args["service"] not in allowed_services:
        raise PermissionError("无权查询该服务")
    start, end = parse_utc(args["start"]), parse_utc(args["end"])
    if not start < end:
        raise ValueError("start 必须早于 end")
    if end - start > max_span:
        raise ValueError("查询跨度超过限制")
    return {**args, "start": start, "end": end}

这里让函数只返回规范化参数,不在校验中顺便发请求,便于直接覆盖非法日期、逆序、超长跨度和越权等边界。执行服务还应按当前身份再次核验权限;发现、校验到真正调用之间,权限可能变化。

校验失败的结果要让调用方知道该修正参数、补充条件还是停止。权限不足不能伪装成“没有日志”,否则模型可能继续换关键词寻找根本不允许访问的内容。

从请求跑到一个明确的结果

把两层组合起来,身份来自应用会话,不能从模型参数里读取:

raw = '{"service":"inventory","start":"2026-10-28T08:00:00Z","end":"2026-10-28T09:00:00Z","limit":20}'
args = decode_query(raw)
query = validate_query(args, allowed_services={"inventory"})
assert query["end"] - query["start"] == timedelta(hours=1)

下载完整校验示例后运行 python query_contract.py。其中的测试把缺字段、多字段、重复字段、布尔值数量、非法日期、无时区时间、逆序区间、超长跨度和越权分别送到对应边界。示例只验证参数,不连接日志后端,也不实现工具协议服务器。

接入工具时,可以将错误映射为稳定的返回类型。下面给出三种应用层响应形状;字段是本例约定,不是某家模型 API 的协议:

[
  {"status":"invalid_arguments","field":"end","reason":"must_follow_start","retryable":false},
  {"status":"forbidden","retryable":false},
  {"status":"ok","items":[],"next_cursor":null,"truncated":false}
]

第一种需要修正请求,第二种需要停止越权调用,第三种才表示查询成功但没有记录。retryable: false 禁止原样自动重试,不代表参数修正后也永远不能发起新请求。权限错误不应暴露调用方无权获知的服务详情。

工具返回也需要契约

查询发出去以后,调用方还需要知道该怎样理解返回值。假如工具只返回一句“没有得到日志”,它可能表示查询成功但没有记录,也可能表示服务暂时不可用。稳定的状态码可以把这些情况分开:成功时给数据和可追踪引用,参数错误时给字段原因,外部状态未知时留下操作标识。

这些结构同样不保证内容正确。分页工具要说明是否还有下一页,截断结果要显式标记,读取到的旧版本数据要带版本或时间。否则模型可能把第一页、截断摘要或过期状态当作完整事实。

超时不等于写入失败

读取日志超时通常可以重试;创建工单、提交任务等写操作则需要先考虑副作用。外部系统可能已完成写入,只是响应没有回到 Harness。

运行时持久化逻辑操作标识后请求外部服务,响应丢失则查询状态,已成功时补记结果,否则保留未知并按后端契约处理。运行时持久化逻辑操作标识后请求外部服务,响应丢失则查询状态,已成功时补记结果,否则保留未知并按后端契约处理。
写操作恢复示意:超时后的查询需要后端支持;无法确认结果时保留 unknown,不推断写入失败。

宽图可横向滚动。查看原图查看原图

幂等键应由运行时为同一个逻辑操作生成、持久保存,重试时复用,并绑定参数摘要。让模型每次重试生成新键,就失去了去重作用。后端的键作用域、保留时间、参数冲突行为也要明确;一个本地 UUID 本身不构成跨系统的幂等保证。

恢复时若查到已成功,就补记结果;若确认未执行且仍允许操作,可以按契约重试;若无法确定,保留 unknown 并进入核对流程。不能把查询接口中的暂时未找到,直接解释为强一致的“没有执行”。

用边界用例检查整条契约

本页可运行示例覆盖参数入口。接入有副作用的工具后,还需在服务替身或隔离环境中注入“写入成功但响应丢失”和重复投递,检查操作记录及外部对象数量。只验证参数函数,无法证明跨系统恢复正确。

这些失败需要不同的预期结果,不能都压成一个 success: false 后无限重试。模型负责提出下一步,运行时负责让这一步满足可执行条件;把条件写成可测试的接口,后续更换模型、提示词或工具实现时才有稳定的回归依据。完整任务的判定方式见 Agent 评估与回归。