图像生成 API 选型：用 JSON Prompt Contract 做安全边界

图像生成 API 选型：用 JSON Prompt Contract 做安全边界 | 前端进阶之旅

Full translation

完整中文译文

短答案：选择你能验证任务生命周期的图像生成 API，然后在它前面放置一个 fail-closed（故障关闭）JSON prompt 契约——前提是该 API 没有独立的审核端点。

聊天模型是一个策略决策点，而非图像被安全创建或持久化存储的证明。因此，我的架构决策是让三个边界保持显式：prompt 准入、生成、发布。每个边界都会产生各自的持久记录和请求标识符。如果某个提供商没有审核端点，应用程序仍然可以在发送 prompt 之前强制执行类型化的安全决策，但必须将该决策视为多个控制手段之一，而非巧妙绕过提供商策略的捷径。

我在这里不信任特性矩阵。它们很少说明超时后会发生什么、成功的响应代表的是接受还是完成、哪个标识符能让运维人员核对缺失的对象。这些细节才是决定哪个 API 适合真实工作负载的关键。

图像生成 API 在聊天模型放行一个 prompt 之前应该证明什么？

我从不变量开始，因为形容词在故障发生时毫无意义。Prompt 只有在策略门返回与应用程序 JSON 契约匹配的文档后才能到达生成阶段。生成的资产只有在其字节、媒体类型、摘要、策略版本和来源记录被持久化之后才能触达用户。重试最多只能为一个逻辑请求创建一个已发布的资产，即使它导致了多次上游尝试。最后，模棱两可的结果在协调机制确定发生了什么之前保持未发布状态。

门控应该返回一个小型的决策对象：allow（是否允许）、stable reason codes（稳定的拒绝原因码）、policy version（策略版本），以及一个 normalized prompt（已批准使用或被忽略的规范化 prompt）。自由格式的解释可以单独记录到日志，但不应该驱动控制流。我还将返回的原因码限制为应用程序自有词汇表内的值。否则，模型可以编造一个令人安心的短语，而调用者可能意外地将其当作许可。

当没有专门的审核端点时，这种分离就变得尤为重要。变通方案不是伪装 prompt 或削弱提供商的控制能力，而是在应用程序层面运行聊天模型检查、本地验证 JSON 结果，并且仍然尊重每个下游的安全响应。如果聊天调用超时、返回格式错误的 JSON、指定了未知原因或省略了策略版本，决策就是拒绝。

生成 API 必须暴露足够的生命周期证据，让调用者能够区分已接受、已完成、已拒绝和状态未知的工作。我寻找的是：有文档化的重试语义、稳定的请求标识符、明确的输出元数据、有界的超时时间，以及一种协调不确定尝试的方式。立即返回图像的响应可以满足这个契约；异步任务也可以。我不会接受的是这样一种集成：HTTP 成功是唯一保留的证据。

失败边界比特性列表更重要

我在一个生成管道中遇到了一次静默故障：调用返回了 200，但副作用从未发生。花了七个小时，一个发布批次才揭示预期对象从未出现在存储中。在此之前，仪表盘看起来很平静，因为它统计的是 HTTP 响应，而工作进程只是把唯一的上游元数据写入了一个临时日志。我们搜索了对象前缀，追溯到发布者的空输入，然后发现我们没有一条持久的关联记录来串联已批准的 prompt、上游请求和预期的对象键。那条缺失的链接比原始响应更重要。团队既无法证明完成，也无法证明安全重试，而从 prompt 回放可能会产生一个重复项，发布者稍后可能会暴露它。修复是操作层面的，而非魔法：在调用前持久化意图、保留上游标识符、在调用后验证对象、仅从已协调状态发布。

那次事件改变了我比较设计的方式。我现在的简短清单是这样的：

没有哪一行能普遍胜出。我不太明白为什么 API 比较常常把这些问题压缩成一个模型质量分数，因为存储和重试边界可能主导用户可见的结果。延迟方面的体验因人而异，但歧义在任何地方都有相同的形态：连接断开后，调用者可能不知道服务器是否采取了行动。RFC 9110 精确地区分了幂等方法，正是因为自动重试并非对每个请求都同样安全。生成调用通常带有类似创建的语义，所以我在重试之前需要提供商明确的文档说明或我自己的去重账本。

把 JSON schema 决策放到关键路径上

下面的 Python 草图将供应商细节隐藏在接口后面。重要的是状态机——记录的意图、严格的决策解析、生成、字节验证和发布——而不是任何虚构的 URL。put_if_absent 边界使发布成为单一的逻辑转换；它的具体一致性和持久化保证仍然需要为你选择的存储系统进行验证。

import hashlib
import json
from dataclasses import dataclass
from typing import Protocol

ALLOWED_REASONS = {"ok", "unsafe_content", "unknown"}

DECISION_SCHEMA = {
    "type": "object",
    "additionalProperties": False,
    "required": ["allow", "reasons", "policy_version", "prompt"],
    "properties": {
        "allow": {"type": "boolean"},
        "reasons": {"type": "array", "items": {"type": "string"}},
        "policy_version": {"type": "string"},
        "prompt": {"type": "string"},
    },
}

class ChatGate(Protocol):
    def decide(self, prompt: str, schema: dict) -> str: ...

class ImageGenerator(Protocol):
    def generate(self, prompt: str, request_id: str) -> bytes: ...

class Ledger(Protocol):
    def record_intent(self, request_id: str, prompt_digest: str) -> None: ...
    def record_denial(self, request_id: str, reasons: list[str]) -> None: ...
    def put_if_absent(self, request_id: str, image: bytes, digest: str) -> bool: ...

@dataclass(frozen=True)
class Decision:
    allow: bool
    reasons: list[str]
    policy_version: str
    prompt: str

def parse_decision(raw: str) -> Decision:
    value = json.loads(raw)
    if set(value) != {"allow", "reasons", "policy_version", "prompt"}:
        raise ValueError("decision fields do not match the contract")
    if type(value["allow"]) is not bool:
        raise ValueError("allow must be boolean")
    if not isinstance(value["reasons"], list) or not all(
        isinstance(reason, str) and reason in ALLOWED_REASONS
        for reason in value["reasons"]
    ):
        raise ValueError("reasons contain an unknown value")
    if not isinstance(value["policy_version"], str) or not value["policy_version"]:
        raise ValueError("policy version is required")
    if not isinstance(value["prompt"], str):
        raise ValueError("prompt must be a string")
    return Decision(**value)

def create_image(
    request_id: str, prompt: str, gate: ChatGate, generator: ImageGenerator, ledger: Ledger
) -> str:
    prompt_digest = hashlib.sha256(prompt.encode()).hexdigest()
    ledger.record_intent(request_id, prompt_digest)

    try:
        decision = parse_decision(gate.decide(prompt, DECISION_SCHEMA))
    except (ValueError, json.JSONDecodeError):
        ledger.record_denial(request_id, ["unknown"])
        return "denied"

    if not decision.allow:
        ledger.record_denial(request_id, decision.reasons)
        return "denied"

    image = generator.generate(decision.prompt, request_id)
    if not image:
        raise ValueError("empty image payload")
    image_digest = hashlib.sha256(image).hexdigest()
    published = ledger.put_if_absent(request_id, image, image_digest)
    return "published" if published else "already_published"

在部署版本中，我还会将策略版本和模型配置绑定到意图记录上、避免将原始 prompt 写入常规日志、验证解码后的媒体文件而非信任文件名，以及使用延迟和拒绝指标作为标签而不将 prompt 文本作为标签。事件回顾中的长篇描述通常始于这些细节被隐式处理的地方。

为什么我拒绝了纯聊天审核的变通方案

我拒绝了一种设计方案：聊天模型返回安全，应用程序就立即暴露图像调用返回的任何结果。它有两个耦合的未知因素：分类器可能产生无效或上下文错误的决策，生成请求可能有模棱两可的结果。布尔值不能识别使用的策略、解释一个稳定的拒绝类别、证明确切批准的 prompt 被生成了，或者确定结果字节被持久化了。如果工程师开始仅仅为了绕过下游控制而重写 prompt，它也会产生策略规避的诱惑。不要那样做。

问题是，我的持久化工作流并非适用于所有系统。对于具有一次性输出的内部原型，同步调用加本地 schema 验证可能是正确的边界；添加队列、账本和协调工作进程几乎不会有收益。当输入语言狭窄且策略可以在没有上下文判断的情况下表达时，使用确定性规则。当提供商的策略正是你需要的策略，并且你乐于将准入与该提供商耦合时，单独使用提供商有文档的安全结果即可。

对于生产选型，我在签字之前运行故障注入测试：格式错误的门控输出、门控超时、接受前后的生成超时、重复的工作进程传递、存储写入冲突，以及对象持久化和发布之间的崩溃。然后我要求团队展示哪些状态是可重试的，哪些需要协调。成本属于该评审范围，但我比较的是完整尝试——聊天决策、生成、重试、存储和运维时间——而不是把每张图像的费用当作架构来对待。

最终决策记录应命名被拒绝的选项及其有效用例、记录所需的一致性和保留属性，并将每条重试规则与文档化的 HTTP 行为关联起来。这不如排行榜那么激动人心。但这也是我如何防止 prompt 安全门变成第二个观察不足的生产系统的方式。

RFC 9110, HTTP Semantics: https://www.rfc-editor.org/rfc/rfc9110