Qora API — AI API Gateway for Developers

AI API Gateway for Developers

One clear API workflow for your apps, scripts and automations.

Output Guardrails: Validating LLM Responses in Production

Layered guardrails validating LLM output before it reaches users

Validate the model’s output because you cannot prove the absence of a failure mode by writing a better instruction. A prompt is a request, not a contract. The same input produces different output across model versions, across providers serving the same weights, and across supposedly identical temperature-0 calls, because batching, kernel selection and floating-point reduction order are not deterministic. Output guardrails are the layer that turns “the model usually does this” into “the system only ever ships this” — and they run after generation, not inside the prompt.

Why you validate the output instead of hardening the prompt

Instruction-following is best-effort. Writing “always return valid JSON” shifts the probability distribution of outputs; it does not create an invariant. That matters because prompt failures are silent: a model that ignores a format instruction returns something plausible until a parser hits it at 3am, whereas a validator fails loudly, at the boundary, with the offending value attached.

Non-determinism has several sources and only one is the temperature parameter. Greedy decoding at temperature 0 breaks ties in the logits, but the logits are a floating-point computation whose reduction order depends on batch size, sequence packing and tensor-parallel layout. Change the batch size and a token ahead by 0.001 logits can flip. Mixture-of-experts routing adds another: a request landing on a different expert set produces a different continuation from the same checkpoint. Version pinning narrows the distribution rather than freezing it, because a pinned snapshot can still be served on different hardware.

So a prompt regression suite tells you what changed, not what is safe right now. Prompts need versioning and review; guardrails are runtime code that needs tests, metrics and an owner. This is about validating what the model says, not what it does — constraining actions is a separate control with a separate threat model, covered in sandboxing AI tool calls.

The validation layers, cheapest first

Run the layers in ascending cost order and stop at the first blocking verdict. A structural failure makes every later check meaningless, and a regex hit is cheaper to act on than a classifier score. Ordering also keeps expensive layers’ error rates out of the picture when a cheap layer already has the answer.

LayerAdded latencyMarginal costWhat it catchesFalse-positive profile
Structural (parse, JSON Schema)0.1-2 msCPU onlyTruncated output, wrong types, missing fields, invented enum values, malformed tool argumentsNear zero if the schema is derived from real payloads; high if the schema is aspirational
Deterministic rules (regex, allow-lists, bounds)under 1 msCPU onlySecret-shaped strings, banned phrases, competitor names, over-length answers, numbers outside a plausible range, tool names not on the allow-listEntirely a function of how precisely the patterns are written; over-broad regex is the largest source of false positives in most stacks
Groundedness (claim vs retrieved context)0.3-2 s with a judge; under 50 ms for the prefilterCheap for substring and embedding lookups, one model call for the residueClaims with no support in the context, and claims the context directly contradictsHigh when retrieval was truncated or the answer legitimately draws on parametric knowledge; needs a prefilter or it blocks good answers
Classifier-based policy checks20-150 msGPU or hosted classification endpoint, priced per requestToxicity, system-prompt self-disclosure, injection payloads echoed back, competitor mentions, PII in free textDominated by the threshold you pick; you cannot reason about it, you have to measure it on labelled data
Human review (sampled)Minutes to hoursHighest, by orders of magnitudeNovel failure modes, calibration of the automated layers, cases where two classifiers disagreeNot applicable, but throughput is capped at tens to hundreds of items per reviewer per day

Two properties matter more than the list. Each layer should return findings rather than a boolean, so the final verdict comes from a policy object instead of being hard-coded inside each check. And the false-positive column decides whether a layer ships, not the true-positive column: a detector that catches every real leak but blocks 8 percent of valid traffic is not a detector, it is an outage.

Schema validation in practice

Make the schema strict enough to be worth running

A schema asserting {"type": "object"} catches nothing. The useful constraints are the boring ones: additionalProperties: false so invented fields surface instead of passing downstream, required on every field the consumer dereferences, enum on anything categorical, and maxItems and maxLength to bound payload size.

Provider strict modes are worth enabling — the strict json_schema response format, Anthropic tool-use schemas, Gemini’s responseSchema — but they remove syntax-level failures, not semantic ones, and they do not survive a failover. A strict schema guarantees the shape of the answer, never its truth. Mechanics are in structured outputs and JSON mode.

A repair loop with a hard attempt cap

When validation fails, the cheapest fix is usually to re-ask the same model with the validator’s error attached. That works often, because the failure is frequently a formatting slip rather than a reasoning failure. It needs a cap: every attempt is a full round trip with full input tokens, so two attempts on a 4,000-token prompt is three times the input cost of a single call.

import json
from dataclasses import dataclass
from typing import Any, Callable

import jsonschema

INVOICE_SCHEMA: dict[str, Any] = {
    "type": "object",
    "additionalProperties": False,
    "required": ["vendor", "currency", "total_cents", "line_items"],
    "properties": {
        "vendor": {"type": "string", "minLength": 1, "maxLength": 200},
        "currency": {"type": "string", "enum": ["USD", "EUR", "GBP"]},
        "total_cents": {"type": "integer", "minimum": 0},
        "line_items": {
            "type": "array",
            "minItems": 1,
            "maxItems": 200,
            "items": {
                "type": "object",
                "additionalProperties": False,
                "required": ["description", "amount_cents"],
                "properties": {
                    "description": {"type": "string", "maxLength": 500},
                    "amount_cents": {"type": "integer", "minimum": 0},
                },
            },
        },
    },
}

MAX_REPAIR_ATTEMPTS = 2

@dataclass(frozen=True)
class SchemaResult:
    ok: bool
    value: dict[str, Any] | None
    attempts: int
    errors: list[str]

def _describe(exc: Exception) -> str:
    if isinstance(exc, jsonschema.ValidationError):
        path = "/".join(str(p) for p in exc.absolute_path) or "<root>"
        return f"{path}: {exc.message}"
    return str(exc)

def parse_with_repair(
    raw: str,
    call_model: Callable[[str, float], str],
    temperature: float = 0.0,
) -> SchemaResult:
    """Validate raw output; on failure re-ask with the validator error attached.

    call_model is injected so this is unit-testable without a network call.
    """
    errors: list[str] = []
    candidate = raw
    for attempt in range(MAX_REPAIR_ATTEMPTS + 1):
        try:
            value = json.loads(candidate)
            jsonschema.validate(value, INVOICE_SCHEMA)
        except (json.JSONDecodeError, jsonschema.ValidationError) as exc:
            errors.append(_describe(exc))
        else:
            return SchemaResult(True, value, attempt, [])

        if attempt == MAX_REPAIR_ATTEMPTS:
            break

        candidate = call_model(
            "Your previous reply failed validation. Fix only the problems listed. "
            "Do not add, remove or reinterpret fields, and do not change a value "
            "that was not listed as invalid.\n"
            "Errors:\n" + "\n".join(f"- {e}" for e in errors[-3:]) + "\n"
            "Return the corrected JSON object and nothing else.",
            temperature,
        )

    return SchemaResult(False, None, MAX_REPAIR_ATTEMPTS, errors)

What the repair loop must never do

The repair prompt above carries a sentence doing real work: do not change a value that was not listed as invalid. Without it, a model asked to fix a schema error will often rewrite the data to make the error disappear — rounding a total so the line items sum, deleting the field with the wrong type, reclassifying a currency to fit the enum. That is data corruption dressed as a successful repair.

Cross-field semantic checks belong outside the repair loop. If your validator asserts that sum(line_items) == total_cents, a repair cannot fix a mismatch, because the model cannot know which side is wrong. Block and route to a human. A repair rate jumping from 2 percent to 15 percent after a prompt edit is the earliest signal of a regression.

Groundedness: unsupported is not the same as contradicted

Extract the atomic claims, then look for support for each in the context you actually retrieved. Three outcomes, and the third is the important one. Supported: a span in the context entails the claim. Unsupported: nothing speaks to it either way — the model may be using parametric knowledge, or retrieval was truncated. Contradicted: the context asserts the opposite. Contradicted claims should be blocked or rewritten; unsupported claims should be downgraded or shown with a citation requirement, because blocking every unsupported claim destroys a summarizer that adds one reasonable inference. See detecting and reducing hallucinations for the taxonomy in depth.

Per-claim judging is quadratic in the obvious implementation, so add a prefilter: normalize each claim, then check for a near-verbatim span, an embedding similarity above a tuned threshold, or a token-overlap ratio. Only the residue goes to a judge, and the judge call is batched — one request containing every unresolved claim, not one per claim.

Worked example. A response contains 40 atomic claims. The prefilter resolves 28 of them, or 70 percent, without a model call. The remaining 12 go to a judge in one batched request returning per-claim entailment labels in roughly 400 ms. Added latency is one round trip, not twelve. Calling the judge once per unresolved claim at 350 ms each would add 4.2 seconds, and the feature would be disabled within a week.

If you use a judge, use a different model than the generator, or at minimum a different prompt with a strict rubric. Ask for a span-level verdict — “quote the sentence that supports this claim, or answer NONE” — rather than a scalar score. A judge forced to produce a citation cannot reward itself with a vague 0.8.

PII and secret leakage in the output

Deterministic detectors cover more than people expect: card numbers with a Luhn check, IBANs with mod-97, national identifier formats, AWS access key prefixes, JWTs by their three-segment structure, PEM private key blocks, connection strings with embedded credentials. Use the checksum wherever the format has one, because a bare digit pattern also matches order numbers.

Regex alone misses the cases that matter: a paraphrased address, a phone number written in words, a name split across a token boundary, an email base64-encoded into a code block, “the card ending in 4242”, or an identifier the model reconstructed from context rather than copied. For free-text names, addresses and locations you need a NER or classifier pass in addition, and it will be the layer with the least predictable error rate.

Redaction versus blocking is a decision about reversibility, not severity. Mask an email as a stable placeholder when the consumer only needs to know an email was present, and keep the mapping server-side. Block when the value is a credential, a regulated identifier, or anything whose exposure is itself the harm. And never log the raw pre-guardrail output: if your guardrail redacts PII and your observability pipeline then stores the original response body, the guardrail has achieved nothing except latency. Log the post-guardrail text plus finding metadata — layer, rule, span offsets, verdict — and put raw output, if you must keep it, in a separate store with short retention and its own audit trail. The wider constraints are in AI data privacy and GDPR.

Fail-open or fail-closed, decided per surface

Fail-open means that when the guardrail itself errors or times out, the output ships. Fail-closed means it does not. The answer is not global, and it is not something you discover in an except block — it is configuration attached to the surface.

Blast radius decides. A support chatbot failing open ships one bad answer to one user who can ask again; the damage is bounded. The same chatbot failing closed turns every classifier timeout into “something went wrong” for the whole user base, an availability incident you caused yourself. A clinical triage flow inverts the calculus: failing open can produce advice that causes physical harm, while failing closed sends the user to a phone number a human answers. Internal analytics fails open by default, because a wrong number in a dashboard is cheap to correct.

Be precise about timeouts. A classifier that times out has said nothing about the content; treating that as “unsafe” produces random outages under load, and load is exactly when timeouts cluster.

Latency budgeting and the streaming problem

Guardrails add round trips, and the budget is tighter than teams assume. Run deterministic checks inline; they cost microseconds. Run the classifier and the groundedness check concurrently, because they are independent, so the wall-clock cost is the maximum rather than the sum.

Worked example. A response is 600 tokens. The PII detector takes 15 ms, the toxicity classifier 90 ms, the policy classifier 120 ms and the groundedness judge 400 ms. Serially that is 625 ms of added latency before the first token reaches the client. With asyncio.gather over the last three it is max(90, 120, 400) = 400 ms. The judge is the critical path, which is why the groundedness prefilter buys more than any micro-optimization of the regexes.

Hold-back windows and retraction UX

Streaming and blocking are in direct tension. Once a token is on the user’s screen you cannot un-show it; you can only append a retraction, and you should assume the user read the original. Three patterns work. First-N-token gating buffers the opening 40 to 80 tokens, validates it, and releases the stream if it passes; most violations are visible in the opening. A sliding hold-back window keeps the last W tokens unreleased while checks run, bounding exposure to the window. Full buffering generates, validates, then renders, which for a 600-token answer at 60 tokens per second moves time-to-first-token from about 0.5 s to about 10 s.

type GateResult = { block: boolean; reason?: string };

export type Gate = {
  holdbackChars: number;
  check: (unreleased: string) => Promise<GateResult>;
};

export async function* guardedStream(
  source: AsyncIterable<string>,
  gate: Gate,
): AsyncGenerator<string> {
  let pending = "";
  let released = 0;

  for await (const chunk of source) {
    pending += chunk;

    // Only inspect text that has not reached the client yet.
    const unreleased = pending.slice(released);
    if (unreleased.length < gate.holdbackChars) continue;

    const { block, reason } = await gate.check(unreleased);
    if (block) {
      yield `\n\n[Response withheld: ${reason ?? "policy"}]`;
      return;
    }

    yield unreleased;
    released = pending.length;
  }

  const tail = pending.slice(released);
  if (tail) {
    const { block, reason } = await gate.check(tail);
    yield block ? `\n\n[Response withheld: ${reason ?? "policy"}]` : tail;
  }
}

The window bounds the text a user can see before a violation is caught, so a retraction undoes a sentence rather than a paragraph. It does not help when the violation is a single token in the middle of a long answer. For high-harm surfaces, accept the latency and buffer the whole response.

Fallback strategies

A blocked response is a routing decision, not an error state. Choose the fallback before you ship the guardrail, because inventing one during an incident produces worse outcomes than a boring canned string.

FallbackUse whenAdded latencyCost per eventWhat the user sees
Safe canned responseThe block was for an out-of-scope or off-policy requestUnder 10 msEffectively zeroA short answer that declines and points to docs or a human
Degrade to a smaller model with a stricter promptThe violation looks prompt-shaped and a constrained model is good enough for the taskOne extra generation, 300 ms to 2 sOne additional generation plus the blocked oneA slightly less rich but valid answer
Hand off to a humanThe domain is high-harm and the request is legitimate but unresolvable automaticallyMinutesHighest, and unbounded per itemA queue position, a callback promise or a ticket number
Structured errorThe caller is a machine, not a personUnder 10 msEffectively zeroA typed error with a retryable flag and the failing rule id

Two rules of thumb. Degrade to a smaller model when the failure came from a permissive prompt, and not when the task is simply hard — a weaker model on a hard task produces a confident wrong answer, which is worse than a block. And never let a fallback re-enter the same guardrail loop without a depth limit, or a persistent violation becomes an infinite regeneration loop.

Measuring a guardrail you intend to ship

Build a labelled set before shipping the layer: a sample of real traffic, human-labelled at the exact verdict granularity you ship, plus a synthetic adversarial set covering the failures you worry about. Then track precision, recall and the false-positive rate on live traffic, split by layer, because a blended number hides which layer is misfiring.

Worked example. You serve 1,000,000 responses a month and the blocking guardrail fires on 0.5 percent of them, so 5,000 blocks. If the measured false-positive rate on known-good outputs is 5 percent, then roughly 0.05 x 995,000 = 49,750 legitimate responses were blocked. You have manufactured an incident an order of magnitude larger than the one you were preventing. You cannot measure that 5 percent from blocked traffic alone — it requires labelling outputs you allowed, which is the step teams skip.

Two more metrics belong on the same dashboard: repair-loop rate, a leading indicator of prompt or model regression, and block rate per rule, because a rule whose block rate is zero is either perfect or broken and it is almost never perfect. Re-run the labelled set on every model version change. Keeping model and prompt changes traceable alongside these numbers is where LLM observability earns its keep. A guardrail nobody measured is one you cannot ship.

A concrete pipeline

The shape that survives production is a set of layer functions returning findings, a policy object resolved per surface, and a verdict computed by severity rather than control flow. The classifier is injected, so the whole thing is testable with a stub.

import asyncio
import re
import time
from dataclasses import dataclass, field
from enum import Enum
from typing import Awaitable, Callable

class Verdict(str, Enum):
    ALLOW = "allow"
    REDACT = "redact"
    BLOCK = "block"
    REVIEW = "review"

@dataclass
class Finding:
    layer: str
    rule: str
    verdict: Verdict
    detail: str

@dataclass
class Outcome:
    verdict: Verdict
    text: str
    findings: list[Finding] = field(default_factory=list)
    timings_ms: dict[str, float] = field(default_factory=dict)

SECRET_PATTERNS = {
    "aws_access_key": re.compile(r"\b(?:AKIA|ASIA)[0-9A-Z]{16}\b"),
    "private_key_block": re.compile(r"-----BEGIN [A-Z ]*PRIVATE KEY-----"),
    "jwt": re.compile(r"\beyJ[A-Za-z0-9_-]{10,}\.[A-Za-z0-9_-]{10,}\.[A-Za-z0-9_-]{10,}\b"),
}

Policy = dict[str, Verdict]

# One explicit policy per surface. Fail-open and fail-closed are data here,
# not an except block somewhere in the request handler.
POLICIES: dict[str, Policy] = {
    "support_chat": {
        "structural": Verdict.BLOCK, "policy": Verdict.BLOCK,
        "grounded": Verdict.REDACT, "on_error": Verdict.ALLOW,
    },
    "clinical_triage": {
        "structural": Verdict.BLOCK, "policy": Verdict.BLOCK,
        "grounded": Verdict.BLOCK, "on_error": Verdict.BLOCK,
    },
    "internal_analytics": {
        "structural": Verdict.REVIEW, "policy": Verdict.REVIEW,
        "grounded": Verdict.ALLOW, "on_error": Verdict.ALLOW,
    },
}

WORST_FIRST = (Verdict.BLOCK, Verdict.REDACT, Verdict.REVIEW, Verdict.ALLOW)

def _worst(findings: list[Finding], fallback: Verdict) -> Verdict:
    for verdict in WORST_FIRST:
        if any(f.verdict is verdict for f in findings):
            return verdict
    return fallback

def redact(text: str, findings: list[Finding]) -> str:
    out = text
    for f in findings:
        if f.layer == "pii" and f.detail:
            out = out.replace(f.detail, f"[REDACTED:{f.rule}]")
    return out

async def run_guardrails(
    text: str,
    surface: str,
    *,
    validate_structure: Callable[[str], list[Finding]],
    classify: Callable[[str], Awaitable[list[Finding]]],
    check_groundedness: Callable[[str], Awaitable[list[Finding]]],
) -> Outcome:
    policy = POLICIES[surface]
    timings: dict[str, float] = {}

    start = time.perf_counter()
    structural = validate_structure(text)
    timings["structural"] = (time.perf_counter() - start) * 1000
    if structural:
        # A structural failure makes every later layer meaningless.
        return Outcome(_worst(structural, policy["structural"]), text, structural, timings)

    start = time.perf_counter()
    rule_findings = [
        Finding("rules", name, Verdict.BLOCK, "secret-shaped string in output")
        for name, pattern in SECRET_PATTERNS.items()
        if pattern.search(text)
    ]
    timings["rules"] = (time.perf_counter() - start) * 1000

    start = time.perf_counter()
    policy_findings, grounded_findings = await asyncio.gather(
        classify(text), check_groundedness(text)
    )
    timings["classifier_and_grounded"] = (time.perf_counter() - start) * 1000

    findings = rule_findings + policy_findings + grounded_findings
    verdict = _worst(findings, Verdict.ALLOW)
    if verdict is Verdict.REDACT:
        text = redact(text, findings)

    return Outcome(verdict, text, findings, timings)

Three choices in that code are load-bearing. Structural failures short-circuit, because there is nothing useful to say about the policy of unparseable output. Classifier and groundedness run under one gather, so added latency tracks the slower of the two. And the error path lives in POLICIES rather than a bare except, so an operator can change it without reading the request handler.

What I would ship first for a customer-facing product

Ship structural validation with a bounded repair loop first, before any classifier. It is free, its false-positive rate is near zero when the schema comes from real payloads, and it eliminates the failures that page you at 3am: truncated JSON, missing fields, wrong types. Instrument the repair rate from day one.

Second, ship deterministic PII and secret detection, fail-closed on secrets and redact on everything else, with raw output kept out of logs from the first commit. It is cheap, defensible in a security review, and the layer most likely to catch a genuine incident.

Third, buffer the full response for the single highest-harm surface, and only that surface. Non-streaming is a real product cost, so spend it where the blast radius justifies it.

Do not ship a policy classifier as a blocker before you have a labelled set and a measured false-positive rate. Do not ship an LLM-as-judge groundedness check as a blocker at all in the first release: run it in shadow mode, log what it would have blocked, and label a few hundred of those decisions by hand. Until you can state your false-positive rate with a number attached, it is a measurement, not a control.

Put model routing behind a gateway rather than wiring providers into application code, so pinning a version, swapping a judge or failing over is a configuration change instead of a deploy.

Frequently asked questions

Do I still need output validation if I use a provider’s strict structured outputs mode?

Yes, for three reasons. Strict modes constrain syntax, not semantics: a strict schema will happily return a well-typed total that does not equal the sum of its line items. They are provider-specific, so the guarantee disappears the moment your gateway fails over. And they say nothing about groundedness, PII or policy.

Should the groundedness judge be a different model from the generator?

Preferably yes, and at minimum it must be a different prompt with an explicit rubric. A judge sharing the generator’s system prompt inherits the same framing and will rationalise the generator’s output. If you can only afford one model, demand a quoted supporting span for every supported verdict.

Can I stream tokens and still enforce guardrails?

Partially. Use a hold-back window so the text you validate has not reached the client, and accept that a late violation still needs a retraction. First-N-token gating covers violations in the opening sentence. Where a leaked token is unacceptable, do not stream.

How do I handle a guardrail false positive in production?

Make it reviewable rather than just blockable. Store the finding, the rule id, the span and a hash of the output, so a reviewer can judge the decision without storing raw text, and give the reviewer an override that feeds back into the labelled set. If one rule accounts for most of your false positives, fix that rule rather than raising the layer threshold.

Do guardrails belong in the client, the gateway, or the application?

Split them by what each knows. The application owns semantic checks, because only it knows the schema, the retrieved context and the policy per surface. The gateway owns the cross-cutting pieces: version pinning, provider failover, redacted logging, rate limits. The client should own nothing that matters for safety.

Conclusion

Output guardrails are the executable specification of what your system is allowed to emit. Order them by cost, make each layer return findings rather than booleans, resolve fail-open versus fail-closed per surface as configuration, and measure the false-positive rate before anything is allowed to block. The layers that ship first are the boring ones: schema validation with a bounded repair loop, deterministic secret and PII detection, and full buffering where a leaked token is unacceptable.

The alternative is not “no guardrails”, it is unmeasured guardrails: regexes nobody tuned, a threshold copied from a blog post, and no idea how many good answers you blocked last month. Routing model traffic through a single OpenAI-compatible endpoint such as qoraapi.com makes the operational side tractable — pinned versions, failover, one place to enforce redaction and logging — but the validation logic is still yours to design.

Related reading

Build AI features with one clear API

Qora API gives you a single, focused gateway to connect your apps, scripts and automations to AI. Start with one request.

qoraapi.com · AI API gateway for developers

Comments

Leave a Reply

Your email address will not be published. Required fields are marked *