DEV Community

YukiKobayashi880
YukiKobayashi880

Posted on

Image API Selection: A JSON Prompt Contract for a Chat Model

Short answer: Choose the image generation API whose job lifecycle you can verify, then put a fail-closed JSON prompt contract in front of it when no separate moderation endpoint exists.

The chat model is a policy decision point, not proof that an image was created safely or stored durably. My architecture decision is therefore to keep three boundaries explicit: prompt admission, generation, and publication. Each boundary emits its own durable record and request identifier. If a provider has no moderation endpoint, the application can still enforce a typed safety decision before sending the prompt, but it must treat that decision as one control among several rather than a clever bypass of provider policy.

I distrust feature matrices here. They rarely say what happens after a timeout, whether a successful response represents acceptance or completion, or which identifier lets an operator reconcile a missing object. Those are the details that determine the best API for a real workload.

What should an image generation API prove before a chat model clears a prompt?

I start with invariants because adjectives don't survive an incident. A prompt may reach generation only after the policy gate returns a document that matches the application's JSON contract. A generated asset may reach a user only after its bytes, media type, digest, policy version, and provenance record have been persisted. A retry may create at most one published asset for one logical request, even if it causes more than one upstream attempt. Finally, an ambiguous result stays unpublished until reconciliation establishes what happened.

The gate should return a small decision object: allow, stable reason codes, a policy version, and a normalized prompt that is either approved for use or ignored. Free-form explanations can be logged separately, but they shouldn't drive control flow. I also cap the returned reason codes to an application-owned vocabulary. Otherwise a model can invent a reassuring phrase that the caller accidentally treats as permission.

This separation matters when there is no dedicated moderation endpoint. The workaround isn't to disguise the prompt or weaken a provider's controls. It is to run an application-level chat model check, validate the JSON result locally, and still honor every downstream safety response. If the chat call times out, returns malformed JSON, names an unknown reason, or omits the policy version, the decision is deny.

Fail closed.

A generation API must then expose enough lifecycle evidence for the caller to distinguish accepted, completed, rejected, and unknown work. I look for documented retry semantics, stable request identifiers, explicit output metadata, bounded timeouts, and a way to reconcile an uncertain attempt. An immediate image response can satisfy that contract; an asynchronous job can too. What I won't accept is an integration where HTTP success is the only evidence retained.

The failure boundary matters more than the feature list

I hit a silent failure in one generation pipeline: the call returned 200, but the side effect never happened. It took seven hours for a publishing batch to reveal that the expected object had never appeared in storage. Until then, the dashboard looked calm because it counted HTTP responses, while the worker had written the only upstream metadata to a transient log. We searched the object prefix, traced the publisher's empty input, and then discovered that we had no durable correlation record joining the approved prompt, the upstream request, and the intended object key. That missing link mattered more than the original response. The team could prove neither completion nor safe retry, and replaying from the prompt risked creating a duplicate that the publisher might later expose. The repair was operational, not magical: persist intent before the call, retain the upstream identifier, verify the object after the call, and publish only from a reconciled state.

That incident changed how I compare designs. My short list now looks like this:

Design Failure isolation Retry posture Main limitation Best fit
Inline chat gate, then synchronous generation Simple request trace; policy and generation remain distinct Retry only with a documented deduplication contract or an application ledger Caller latency includes both decisions Low-volume interactive flows with bounded generation time
Durable intent, chat gate, then queued generation Each transition is recorded and replayable Worker retries from recorded state; publication is deduplicated locally More storage, queueing, and reconciliation work Production pipelines where lost or duplicate assets matter
Provider-native safety response only Few moving parts Entire retry policy follows one API contract Application-specific policy can't be expressed independently Prototypes or workloads whose policy exactly matches the documented service boundary
Local deterministic rules before generation Predictable and cheap to execute Easy to repeat Weak on contextual language and costly to maintain as the policy grows Narrow vocabularies with well-defined prohibited terms

No row wins universally. I'm not sure why API comparisons so often collapse these into a model-quality score, because the storage and retry boundaries can dominate the user-visible outcome. Your mileage may vary on latency, but ambiguity has the same shape everywhere: after a broken connection, the caller may not know whether the server acted. RFC 9110 distinguishes idempotent methods precisely because automatic retry is not equally safe for every request. A generation call commonly carries create-like semantics, so I require explicit provider documentation or my own deduplication ledger before retrying it.

No guesswork.

Put the JSON schema decision on the critical path

The following Python sketch keeps vendor details behind interfaces. The important part is the state machine β€” recorded intent, strict decision parsing, generation, byte verification, and publication β€” rather than any invented URL. The put_if_absent boundary makes publication a single logical transition; its concrete consistency and durability guarantees still need to be verified for the storage system you choose.

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"
Enter fullscreen mode Exit fullscreen mode

In a deployed version, I would also bind the policy version and model configuration to the intent record, keep raw prompts out of routine logs, validate the decoded media rather than trusting a filename, and emit latency and denial metrics without using prompt text as a label. The long paragraph in an incident review usually starts where those details were left implicit.

Why I rejected a chat-only moderation workaround

I rejected the design in which a chat model returns safe and the application immediately exposes whatever the image call returns. It has two coupled unknowns: the classifier can produce an invalid or contextually wrong decision, and the generation request can have an ambiguous outcome. A boolean doesn't identify the policy used, explain a stable denial category, prove that the exact approved prompt was generated, or establish that the resulting bytes were durably stored. It also creates a policy-evasion temptation if engineers begin rewriting prompts merely to get past downstream controls. Don't do that.

The catch is that my durable workflow isn't suitable for every system. For an internal prototype with disposable output, a synchronous call plus local schema validation may be the right boundary; adding a queue, ledger, and reconciliation worker would buy little. Stick with deterministic rules when the input language is narrow and policy can be expressed without contextual judgment. Use a provider's documented safety result alone when its policy is exactly the policy you need and you are comfortable coupling admission to that provider.

For production selection, I run fault-injection tests before signing off: malformed gate output, gate timeout, generation timeout before and after acceptance, duplicate worker delivery, storage write conflict, and a crash between object persistence and publication. I then ask the team to show which states are retryable and which need reconciliation. Cost belongs in that review, but I compare complete attempts β€” chat decision, generation, retries, storage, and operator time β€” rather than treating a per-image figure as the architecture.

The final decision record should name the rejected option and its valid use case, record the required consistency and retention properties, and link each retry rule to documented HTTP behavior. That's less exciting than a leaderboard. It is also how I keep a prompt-safety gate from becoming a second, poorly observed production system.

References

Top comments (0)