Structured Output Generation

Your CI incident triage pipeline ingests hundreds of raw build failures every minute. A runner reports: "Run RUN-842 failed in unit-tests after auth-cache timed out." Downstream services need a typed payload containing the run identifier and an allowed status value. That record can start a review or telemetry lookup. It doesn't, by itself, justify a rollback.

Prompting an unconstrained chat model with "Extract the fields and respond in JSON" can break that interface. A reply such as Sure! Here is the data: { 'run_id': 'RUN-842' } fails json.loads on its leading text; single quotes would also violate JSON syntax. Other replies may use Markdown fences, invent a status such as "exploded", or omit "run_id".

Autoregressive models predict next tokens rather than making type-safe remote procedure calls. One way to bridge that gap is structured output generation through grammar-guided decoding. Before sampling each token, the runtime masks continuations outside its compiled grammar.^{[1]Reference 1Efficient Guided Generation for Large Language Models.https://arxiv.org/abs/2307.09702} The supported grammar and the application's complete contract aren't always the same.

This step isolates one decoding cycle on a four-token toy vocabulary. Only { is legal in this fixture. A real tokenizer may also offer legal whitespace or tokens that start with { and continue into a key. Vocabularies vary by model, and tokens can contain multiple punctuation marks or partial words. We'll trace how runtimes enforce a supported grammar, then check what that grammar leaves to the application.

Why "please return JSON" isn't enough

Sending a polite formatting prompt to an LLM leaves structural validation entirely to chance:

prompt = """
Extract run_id and status from this build log:
"Run RUN-842 failed in unit-tests after auth-cache timed out."
Return JSON.
"""

Human readers understand the instructions immediately, but an unconstrained model can fail in several distinct ways:

The parser can't infer whether a malformed reply arose from formatting habits, ambiguous instructions, or another model error. Its job is to enforce the interface. A formatting request alone doesn't provide that boundary.

Reliable systems separate requesting a format from enforcing a contract. These interfaces constrain different parts of a response; they aren't five successive levels every application must climb:

Prompt-only formatting is a gamble. You can write regex cleaners, JSON repair wrappers, and retry loops, but your application code remains vulnerable to unexpected edge cases.

JSON mode doesn't enforce your fields. A completed JSON response such as {"author": "alice"} can satisfy it when you expected run_id and status.^{[2]Reference 2Introducing Structured Outputs in the APIhttps://openai.com/index/introducing-structured-outputs-in-the-api/} JSON itself also permits arrays and scalar values; your application still needs an object-root contract. Check the provider's completion and refusal states before parsing.

Constrained decoding moves enforcement into generation: the runtime blocks continuations outside its compiled language.^{[1]Reference 1Efficient Guided Generation for Large Language Models.https://arxiv.org/abs/2307.09702} This covers only supported JSON Schema constraints. For example, llama.cpp's converter can silently skip unsupported features, so grammar-valid output isn't necessarily valid against the original schema.^{[3]Reference 3llama.cpp Grammars Documentationhttps://github.com/ggml-org/llama.cpp/blob/master/grammars/README.md} Completion, client validation, and provider-specific exceptions remain separate checks.

As checked on October 10, 2026, Claude has native JSON outputs through output_config.format as well as strict tool use. Ordinary text mode isn't a JSON guarantee. Claude's documentation also lists exceptions including refusal, token exhaustion, and string enum capitalization differences. If your contract requires exact enum spelling, reject a mismatch or use an explicit, unambiguous application mapping; don't silently accept arbitrary strings.^{[4]Reference 4Structured outputshttps://platform.claude.com/docs/en/build-with-claude/structured-outputs} Gemini likewise supports a schema subset and calls for application validation of values.^{[5]Reference 5Structured outputshttps://ai.google.dev/gemini-api/docs/structured-output}

The script below demonstrates why syntax validation alone is insufficient for typed interfaces:

import json
import math

REQUIRED = ("run_id", "status")
STATUSES = {"queued", "passed", "failed", "unknown"}

def reject_constant(value):
    raise ValueError(f"nonfinite JSON number: {value}")

def finite_float(value):
    result = float(value)
    if not math.isfinite(result):
        raise ValueError("JSON number exceeds finite float range")
    return result

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

def load_record(raw: str):
    return json.loads(
        raw,
        object_pairs_hook=unique_object,
        parse_constant=reject_constant,
        parse_float=finite_float,
    )

def validate_triage(payload: object) -> dict[str, str]:
    if not isinstance(payload, dict):
        raise TypeError("root must be an object")
    extra = sorted(set(payload) - set(REQUIRED))
    missing = [key for key in REQUIRED if key not in payload]
    if extra or missing:
        raise ValueError(f"shape mismatch extra={extra} missing={missing}")
    record: dict[str, str] = {}
    for key in REQUIRED:
        value = payload[key]
        if not isinstance(value, str) or not value.strip():
            raise TypeError(f"{key} must be a nonempty string")
        record[key] = value
    if record["status"] not in STATUSES:
        raise ValueError(f"unsupported status enum: {record['status']}")
    return record

syntax_valid_wrong_shape = load_record('{"foo": "bar"}')
schema_valid = load_record('{"run_id": "RUN-842", "status": "failed"}')

try:
    validate_triage(syntax_valid_wrong_shape)
except ValueError:
    print("JSON-only result: rejected by schema")

print("structured record:", validate_triage(schema_valid))

for raw in ['{"run_id":"RUN-842","run_id":"RUN-999"}', '{"status":NaN}']:
    try:
        load_record(raw)
    except ValueError as error:
        print("parser rejected:", error)

JSON-only result: rejected by schema
structured record: {'run_id': 'RUN-842', 'status': 'failed'}
parser rejected: duplicate JSON key: run_id
parser rejected: nonfinite JSON number: NaN

Python's standard json.loads silently overwrites duplicate keys (keeping the last value) and parses non-standard constants like NaN and Infinity.^{[6]Reference 6json: JSON encoder and decoderhttps://docs.python.org/3/library/json.html} In production pipelines, accepting duplicate keys or nonfinite floats creates silent corruption. Our custom parser hooks reject them explicitly.

How grammar-guided decoding works

Instead of letting the model produce arbitrary text and validating it afterwards, constrained decoders enforce rules during autoregressive generation.

The backend compiles a reusable grammar representation. At each step, it derives a token mask from the current parser state and applies it before sampling. After accepting a token, it advances that state. An accepting state permits termination; it doesn't always force it, because trailing whitespace or further content may still be legal. Token budgets can interrupt generation before a complete value exists.

From schema to logit masking

The mathematical mechanism works by directly modifying the unnormalized logit vector before the softmax operation.

1. State representation: Let $S_t$ represent the parser's state at decoding step $t$. For regular expressions and flat enums, $S_t$ is a state in a Deterministic Finite Automaton (DFA). For arbitrarily recursive JSON structures, a stack-based parser also tracks unresolved open braces and brackets. Fixed-depth schemas can use finite-state representations instead.
2. Legal token subset: The engine determines the set of legal vocabulary token IDs: $\mathcal{V}_{\text{valid}}(S_t) \subseteq \mathcal{V}$ Each ordinary token in this set must leave a prefix that can still reach a valid completion. Special end tokens need their own acceptance rule.
3. Logit masking: Let $z_t \in \mathbb{R}^{|\mathcal{V}|}$ denote the raw output logits from the transformer's language model head. We construct a mask vector $m_t \in \{0, -\infty\}^{|\mathcal{V}|}$: $m_t(v) = \begin{cases} 0 & \text{if } v \in \mathcal{V}_{\text{valid}}(S_t) \\ -\infty & \text{if } v \notin \mathcal{V}_{\text{valid}}(S_t) \end{cases}$
4. Renormalized sampling: The masked logits $z'_t = z_t + m_t$ pass through softmax: $P(y_t = v \mid x_{<t}) = \frac{\exp(z'_t(v))}{\sum_{u \in \mathcal{V}} \exp(z'_t(u))} = \begin{cases} \frac{\exp(z_t(v))}{\sum_{u \in \mathcal{V}_{\text{valid}}(S_t)} \exp(z_t(u))} & \text{if } v \in \mathcal{V}_{\text{valid}}(S_t) \\ 0 & \text{otherwise} \end{cases}$

Under this mask, illegal tokens have zero probability. The equation assumes temperature-one sampling and a nonempty legal set with finite scores. If no representable legal continuation remains, stop with an error; don't remove the grammar to make sampling succeed.

Here is a minimal implementation of this masking operation:

import math

NEG_INF = float("-inf")

def mask_invalid_tokens(logits: dict[str, float], valid_tokens: set[str]) -> dict[str, float]:
    return {
        token: score if token in valid_tokens else NEG_INF
        for token, score in logits.items()
    }

def softmax(logits: dict[str, float]) -> dict[str, float]:
    if any(not math.isfinite(score) and score != NEG_INF for score in logits.values()):
        raise ValueError("logits must be finite or negative infinity")
    finite = {token: score for token, score in logits.items() if score != NEG_INF}
    if not finite:
        raise ValueError("no legal token with a finite score")
    peak = max(finite.values())
    weights = {token: math.exp(score - peak) for token, score in finite.items()}
    total = sum(weights.values())
    return {token: (weights[token] / total if token in weights else 0.0) for token in logits}

next_token_logits = {
    "The": 0.45,
    "{": 0.30,
    "Sure": 0.15,
    "JSON": 0.10,
}

masked = mask_invalid_tokens(next_token_logits, valid_tokens={"{"})
probs = softmax(masked)
chosen = max(probs, key=probs.get)

print("before softmax (%):", {token: round(prob * 100, 1) for token, prob in softmax(next_token_logits).items()})
print("masked logits:", {token: masked[token] for token in masked})
print("softmax:", {token: round(probs[token], 3) for token in probs})
print("chosen token:", chosen)
assert chosen == "{"
assert probs["{"] == 1.0

before softmax (%): {'The': 30.2, '{': 26.0, 'Sure': 22.4, 'JSON': 21.3}
masked logits: {'The': -inf, '{': 0.3, 'Sure': -inf, 'JSON': -inf}
softmax: {'The': 0.0, '{': 1.0, 'Sure': 0.0, 'JSON': 0.0}
chosen token: {

Even though "The" had the highest initial logit (0.45 vs 0.30), masking forces its logit to $-\infty$. The sampling distribution collapses entirely onto the only valid token: "{".

The tokenizer alignment problem

Implementing constrained decoding in practice reveals an important mismatch: grammars operate on characters or bytes, while language models operate on subword tokens.

A single vocabulary token can contain multiple characters (like ","status":"failed"), or a single character can span multiple UTF-8 byte tokens. The decoding engine can't just check whether the first character of a token is allowed. It must evaluate whether the entire token forms a valid path in the automaton without overshooting into an invalid state.^{[1]Reference 1Efficient Guided Generation for Large Language Models.https://arxiv.org/abs/2307.09702}

Consider an automaton in a state expecting the status enum value "failed".

• Token fa is valid: it advances the state toward "failed".
• Token failed" is valid: it completes the enum, consumes the closing quote, and moves to whatever delimiter the enclosing object permits next.
• Token failed"! is invalid: it completes the enum, but immediately appends an illegal exclamation mark. Even though its prefix is valid, accepting this token would drive the parser into an unrecoverable error state. The engine must mask this token out.

Because evaluating thousands of tokens against an automaton on every forward pass would destroy inference throughput, modern engines use specialized indexing techniques:

• Indexed finite-state decoding: The approach in Willard and Louf's 2023 paper precomputes allowed token IDs for each reachable DFA state. Looking up that index avoids reparsing the whole vocabulary; constructing, storing, and applying masks still has costs.^{[1]Reference 1Efficient Guided Generation for Large Language Models.https://arxiv.org/abs/2307.09702}
• llguidance: Uses an Earley context-free parser, a regex-based lexer, and a tokenizer trie to compute masks.^{[7]Reference 7llguidance: Super-fast Structured Outputshttps://github.com/guidance-ai/llguidance}
• XGrammar: Partitions the vocabulary into context-independent tokens (whose validity depends only on the current grammar rule, precomputed as static bitmasks) and context-dependent tokens (which require dynamic stack evaluation). This drastically reduces per-step CPU work.^{[8]Reference 8XGrammar: Flexible and Efficient Structured Generation Engine for Large Language Modelshttps://arxiv.org/abs/2411.15100}

Regular expressions vs. context-free grammars

Choosing the right automaton representation depends on the structural complexity of your schema:

• Deterministic Finite Automata (DFAs): Suitable for regular patterns such as closed string enums and specified identifier formats. For a fixed automaton, tracking its current state uses constant working memory with respect to output length. The automaton and token index can still grow substantially with the grammar.
• Context-Free Grammars (CFGs) / stack-based parsers: Handle arbitrary recursive nesting, such as JSON trees whose children can contain more children. Matching delimiters requires remembering unresolved nesting. A CFG handles syntax; it doesn't automatically enforce every JSON Schema constraint or cross-field relationship.

Bounded-depth JSON syntax can be represented by a finite-state machine, although that representation may be large. Unbounded length isn't unbounded nesting: a list of any number of fixed-shape objects can be regular. For example, [], [{}], and [{},{},{}] fit \[(\{\}(,\{\})*)?\] regardless of the number of empty objects. Allow those objects to contain arbitrarily nested objects, and a simple DFA is no longer sufficient.

Run a miniature constrained decoder

To see how whole-token validation and sequence termination work together, here is a complete decoder for a two-sentence language: {"status":"passed"} and {"status":"failed"}.

Notice how token iled"}! is rejected because it overshoots the grammar, and how <eos> is masked out until the object is completely closed:

LANGUAGE = ('{"status":"passed"}', '{"status":"failed"}')
VOCAB = ('The', '{"status":"', 'fa', 'passed', 'iled"}', 'iled"}!', '"}', '<eos>')
SCORES = {
    'The': 9.0,
    '{"status":"': 1.0,
    'fa': 3.0,
    'passed': 2.0,
    'iled"}': 1.0,
    'iled"}!': 8.0,
    '"}': 1.0,
    '<eos>': 7.0,
}

def allowed(prefix: str, token: str) -> bool:
    if token == '<eos>':
        return prefix in LANGUAGE
    if not token:
        return False
    candidate = prefix + token
    return any(record.startswith(candidate) for record in LANGUAGE)

def decode(max_steps=8):
    prefix, trace = '', []
    for _ in range(max_steps):
        legal = [token for token in VOCAB if allowed(prefix, token)]
        if not legal:
            raise ValueError('grammar has no representable continuation')
        token = max(legal, key=lambda item: SCORES[item])
        trace.append(token)
        if token == '<eos>':
            return prefix, trace
        prefix += token
    raise ValueError('token budget ended before complete output')

record, trace = decode()
print('tokens:', trace)
print('record:', record)
print('overshooting token legal:', allowed('{"status":"fa', 'iled"}!'))
assert record in LANGUAGE
assert not allowed('{"status":"fa', '<eos>')

tokens: ['{"status":"', 'fa', 'iled"}', '<eos>']
record: {"status":"failed"}
overshooting token legal: False

The decoder successfully produces {"status":"failed"} despite 'The' and 'iled"}!' having higher unconstrained scores. It prevents early stopping by withholding <eos> until the string matches a complete sentence in the language.

Libraries and serving stacks

Depending on whether you manage local GPU inference or call hosted provider APIs, different tools handle grammar compilation and logit masking.

Outlines

Outlines provides model adapters and a reusable Generator. In its current v1 API, the generator returns raw text, including for a Pydantic output type. Parse that text into a validated instance yourself:^{[9]Reference 9Generator APIhttps://dottxt-ai.github.io/outlines/latest/features/core/generator/}

from typing import Literal

import outlines
from outlines import Generator
from pydantic import BaseModel, ConfigDict
from transformers import AutoModelForCausalLM, AutoTokenizer

class RunTriage(BaseModel):
    model_config = ConfigDict(extra="forbid", strict=True)
    run_id: str
    status: Literal["queued", "passed", "failed", "unknown"]
    failing_job: str | None

model_name = "HuggingFaceTB/SmolLM2-135M-Instruct"
model = outlines.from_transformers(
    AutoModelForCausalLM.from_pretrained(model_name),
    AutoTokenizer.from_pretrained(model_name),
)

triage_generator = Generator(model, RunTriage)
raw = triage_generator(
    "Extract run_id, status (queued/passed/failed/unknown), and failing_job (null if absent): "
    "Run RUN-842 failed in unit-tests after auth-cache timed out.",
    max_new_tokens=120,
)
triage = RunTriage.model_validate_json(raw)

Reuse the generator where its backend permits reusable compilation artifacts; don't assume every adapter uses a DFA. Outlines supports outlines_core, llguidance, and XGrammar backends, with different model and output-type compatibility.^{[10]Reference 10Structured Generation Backendshttps://dottxt-ai.github.io/outlines/latest/features/advanced/backends/} Measure cold and warm requests separately. This model-download example isn't executed in the CPU fixtures below.

llama.cpp

In local edge deployments or C++ runtimes, llama.cpp uses GBNF (GGML BNF) grammars to constrain sampling.^{[11]Reference 11llama.cpp: Inference of LLaMA model in pure C/C++https://github.com/ggml-org/llama.cpp[3]Reference 3llama.cpp Grammars Documentationhttps://github.com/ggml-org/llama.cpp/blob/master/grammars/README.md} The runtime converts a subset of JSON Schema into GBNF rules.

Be aware that unsupported schema features may be ignored without warning during GBNF conversion. You should inspect the generated GBNF rules and continue to run client-side Pydantic validation on the output. llama.cpp doesn't insert schema definitions into the system prompt automatically, so your prompt must describe the expected fields and semantics explicitly.^{[3]Reference 3llama.cpp Grammars Documentationhttps://github.com/ggml-org/llama.cpp/blob/master/grammars/README.md}

Serving stacks: SGLang, vLLM, and XGrammar

In production serving clusters, high-concurrency engines integrate grammar masking directly into the batch scheduler.

SGLang's original paper describes two distinct optimizations:^{[12]Reference 12SGLang: Efficient Execution of Structured Language Model Programs.https://arxiv.org/abs/2312.07104}

1. RadixAttention: Reuses the KV cache across requests that share system prompts or schema prefixes, reducing Time To First Token (TTFT).
2. Compressed finite state machines: Merge paths with forced string continuations so the runtime can jump forward over known text, skipping some token-by-token decode steps. Retokenization keeps the appended text aligned with the model's tokenizer. The gain comes from skipping decode steps, with retokenization overhead.

That historical design isn't the whole current serving stack. The current documentation lists XGrammar as the default grammar backend, with Outlines and llguidance alternatives.^{[13]Reference 13SGLang Structured Outputs Documentationhttps://docs.sglang.io/docs/advanced_features/structured_outputs}

vLLM accepts a structured_outputs parameter block and supports XGrammar and Guidance/llguidance backends. Its legacy guided_json, guided_regex, and related fields were removed in v0.12.0, rather than remaining merely deprecated. Use the current request fields and check the selected backend's capabilities.^{[14]Reference 14Structured Outputshttps://docs.vllm.ai/en/latest/features/structured_outputs/}

XGrammar precomputes context-independent token masks and checks stack-dependent cases dynamically.^{[8]Reference 8XGrammar: Flexible and Efficient Structured Generation Engine for Large Language Modelshttps://arxiv.org/abs/2411.15100} Given the same previously accepted tokens, CPU mask generation and the GPU forward pass can run concurrently; they synchronize before sampling the next token. The mask can't be prepared from a future token that hasn't been sampled. Overlap can hide mask costs when they fit inside GPU work; it isn't a universal zero-overhead guarantee.

Function calling vs. structured outputs

The same JSON Schema can be enforced through two different interaction patterns. Knowing when to use each prevents unnecessary architecture complexity.

When to use function calling

Function calling fits interactive agents that must inspect or modify external systems. The model decides which tool to invoke, the host application executes the action with appropriate authorization checks, and the results are fed back into the context window for a final response.

Strict function calling ensures that the generated tool arguments strictly match the tool's parameter schema:

tools = [
    {
        "type": "function",
        "name": "get_run_status",
        "description": "Fetch current CI status for a run.",
        "strict": True,
        "parameters": {
            "type": "object",
            "properties": {
                "run_id": {"type": "string"}
            },
            "required": ["run_id"],
            "additionalProperties": False
        }
    }
]

tool_schema = tools[0]
example_routes = {
    "What happened to RUN-842?": "get_run_status",
    "Hi!": "text_response",
}
print("tool:", tool_schema["name"])
print("strict:", tool_schema["strict"])
print("routes:", example_routes)

tool: get_run_status
strict: True
routes: {'What happened to RUN-842?': 'get_run_status', 'Hi!': 'text_response'}

This is an OpenAI Responses-style tool definition. Enabled strict mode constrains supported argument shapes; it doesn't authorize execution. The printed routes are a hand-written fixture, not a model's observed tool choices. Validate arguments and host policy before running a tool.

When to use structured outputs

Structured outputs fit data extraction, document classification, and pipeline stages where the model's output is consumed directly by application code.

In Pydantic, extra="forbid" makes the local validator reject unexpected properties and adds a closed-object constraint to its schema. It doesn't control a model unless a generation backend receives and enforces that schema. In this example, failing_job: str | None is required but nullable because it has no default:^{[15]Reference 15Strict Modehttps://pydantic.dev/docs/validation/latest/concepts/strict_mode/}

from typing import Literal
from pydantic import BaseModel, ConfigDict, ValidationError

class RunStatusUpdate(BaseModel):
    model_config = ConfigDict(extra="forbid", strict=True)
    run_id: str
    status: Literal["queued", "passed", "failed", "unknown"]
    failing_job: str | None

schema = RunStatusUpdate.model_json_schema()
payload = {"run_id": "RUN-842", "status": "failed", "failing_job": None}
update = RunStatusUpdate.model_validate(payload)

print("required:", schema["required"])
print("closed object:", schema["additionalProperties"] is False)
print("nullable job:", update.failing_job)

try:
    RunStatusUpdate.model_validate({"run_id": "RUN-842", "status": "failed"})
except ValidationError:
    print("omitted job: rejected")

required: ['run_id', 'status', 'failing_job']
closed object: True
nullable job: None
omitted job: rejected

In OpenAI's strict mode, all fields in the schema must be marked as required, and additionalProperties must be set to False.^{[16]Reference 16Structured outputshttps://developers.openai.com/api/docs/guides/structured-outputs} An optional property must be defined as a required field whose type allows null.

A local Pydantic model can enforce constraints that the generation backend never received. For example, Claude's SDK helpers can remove unsupported numeric bounds from the transmitted schema, describe the bounds in field descriptions, and validate against the original model after generation.^{[4]Reference 4Structured outputshttps://platform.claude.com/docs/en/build-with-claude/structured-outputs} A description guides the model; it isn't the missing grammar rule.

This CPU fixture manually constructs a simplified wire schema. Predict whether the candidate retry_count: -1 passes the original local contract. It doesn't call a provider or reproduce an SDK's complete transformation algorithm.

from copy import deepcopy
from pydantic import Field

class RetrySuggestion(BaseModel):
    model_config = ConfigDict(extra="forbid", strict=True)
    retry_count: int = Field(ge=0)

original_schema = RetrySuggestion.model_json_schema()
wire_schema = deepcopy(original_schema)
wire_schema["properties"]["retry_count"].pop("minimum")
wire_schema["properties"]["retry_count"]["description"] = "Must be at least 0"
candidate = {"retry_count": -1}

print("local minimum:", original_schema["properties"]["retry_count"]["minimum"])
print("wire minimum present:", "minimum" in wire_schema["properties"]["retry_count"])
print("candidate fits wire integer shape:", type(candidate["retry_count"]) is int)
try:
    RetrySuggestion.model_validate(candidate)
except ValidationError as error:
    print("local rejection:", error.errors()[0]["type"])

local minimum: 0
wire minimum present: False
candidate fits wire integer shape: True
local rejection: greater_than_equal

Inspect the schema actually sent by your adapter and retain the original validator. Provider support for a keyword can differ by endpoint or model; an SDK-generated schema isn't automatically identical to the local contract.

Here is how you call OpenAI's Structured Outputs endpoint with a Pydantic model:

import os
from openai import OpenAI

client = OpenAI()
response = client.responses.parse(
    model=os.environ["OPENAI_STRUCTURED_MODEL"],
    input="Run RUN-842 failed in unit-tests.",
    text_format=RunStatusUpdate,
)
update = response.output_parsed
if response.status != "completed":
    raise RuntimeError(f"incomplete response: {response.incomplete_details}")
if any(part.type == "refusal" for item in response.output if item.type == "message"
       for part in item.content):
    raise RuntimeError("provider refusal")
if update is None:
    raise RuntimeError("no parsed record")

The SDK may also raise a parsing or validation exception; handle that separately from transport failures. Check the completion state, refusals, and presence of parsed data before business logic. For streaming, a delta or partially parsed object isn't a committed record. Buffer the final candidate and apply the complete contract and host checks before acting on it.

The script below demonstrates how to separate shape validation from authorization and ground-truth verification:

CI = {"RUN-842": {"project": "project-alpha", "status": "failed", "failing_job": "unit-tests"}}

def validate_update(payload: object) -> dict[str, str]:
    return validate_triage(payload)

def decide_update(allowed_projects: set[str], update: dict[str, str]) -> str:
    trusted = CI.get(update["run_id"])
    if trusted is None or trusted["project"] not in allowed_projects:
        return "deny"  # Don't leak whether an unauthorized run ID exists
    if update["status"] != trusted["status"]:
        return "review: contradicts CI"
    return "show"

parsed = validate_update({"run_id": "RUN-842", "status": "failed"})
print("format gate:", parsed)
print("authorized and true:", decide_update({"project-alpha"}, parsed))
print("other project:", decide_update({"project-beta"}, parsed))

false_claim = validate_update({"run_id": "RUN-842", "status": "passed"})
print("authorized but false:", decide_update({"project-alpha"}, false_claim))

format gate: {'run_id': 'RUN-842', 'status': 'failed'}
authorized and true: show
other project: deny
authorized but false: review: contradicts CI

The model can produce a perfectly valid JSON object claiming that RUN-842 passed, but application logic catches the factual contradiction against the CI database. Valid shape doesn't guarantee truthful content.

Systems trade-offs: latency, throughput, and schema design

Enforcing grammar constraints introduces real systems overhead. Understanding where this latency occurs lets you design schemas that maintain high serving throughput.

Where latency originates

Systems costs fall into distinct phases across the request lifecycle:

The quality-compliance trade-off

Tam et al. (2024) found worse results under some formatting restrictions on their reasoning tasks, while classification results sometimes improved. The study includes prompt-based formats and JSON mode on models of that period; it isn't proof that every current schema-constrained model loses reasoning ability.^{[17]Reference 17Let Me Speak Freely? A Study on the Impact of Format Restrictions on Performance of Large Language Modelshttps://arxiv.org/abs/2408.02442}

A restricted final-response format doesn't necessarily restrict the model's internal reasoning channel. Hosted reasoning models can spend reasoning tokens before producing their visible answer.^{[18]Reference 18Reasoning modelshttps://developers.openai.com/api/docs/guides/reasoning}

One design worth evaluating is to put a short source quote before a decision. It gives the application something concrete to check. JSON Schema itself doesn't impose object-key order, so verify whether your backend preserves the desired ordering. Treat any quality improvement as an evaluation question, not a guarantee:

import re

ALLOWED_DECISIONS = {"retry_tests", "rollback", "escalate", "unknown"}
SOURCE_LOG = "Run RUN-842 failed in unit-tests after auth-cache timed out."

def validate_decision(payload: object, source: str) -> dict[str, str]:
    if not isinstance(payload, dict):
        raise TypeError("root must be an object")
    required = ("evidence_quote", "run_id", "decision")
    extra = sorted(set(payload) - set(required))
    missing = [key for key in required if key not in payload]
    if extra or missing:
        raise ValueError(f"shape mismatch extra={extra} missing={missing}")
    quote = payload["evidence_quote"]
    run_id = payload["run_id"]
    decision = payload["decision"]
    if not isinstance(quote, str) or not isinstance(run_id, str):
        raise TypeError("evidence_quote and run_id must be strings")
    if not isinstance(decision, str) or decision not in ALLOWED_DECISIONS:
        raise ValueError("unknown decision")
    if not quote.strip() or quote not in source:
        raise ValueError("quote is not an exact span of the supplied source")
    if not run_id.strip():
        raise ValueError("empty run ID")
    if not re.search(r"(?<![\w-])" + re.escape(run_id) + r"(?![\w-])", quote):
        raise ValueError("run ID does not appear in evidence")
    return {"evidence_quote": quote, "run_id": run_id, "decision": decision}

decision = validate_decision(
    {
        "evidence_quote": "Run RUN-842 failed in unit-tests after auth-cache timed out.",
        "run_id": "RUN-842",
        "decision": "retry_tests",
    },
    SOURCE_LOG,
)
print("decision:", decision["decision"])
print("quote occurs in source:", decision["evidence_quote"] in SOURCE_LOG)

try:
    validate_decision({**decision, "evidence_quote": "RUN-842 needs an immediate rollback."}, SOURCE_LOG)
except ValueError as error:
    print("invented quote:", error)

decision: retry_tests
quote occurs in source: True
invented quote: quote is not an exact span of the supplied source

This fixture validates a supplied quote; it doesn't call a model or measure reasoning quality. An exact span plus a matching run ID establishes source membership, not support for the proposed action. Predict the result if we keep the same true quote and replace retry_tests with rollback:

unsupported = validate_decision({**decision, "decision": "rollback"}, SOURCE_LOG)
print("quote validator accepted:", unsupported["decision"])
print("rollback justification checked:", False)
assert unsupported["evidence_quote"] == decision["evidence_quote"]

quote validator accepted: rollback
rollback justification checked: False

The log says tests failed after a timeout. It supplies neither a deployed release to undo nor evidence that rollback would help. A separate action policy needs current system state, evidence relevant to that action, and execution authority.

Flatten nested structures

Flattening is an option when a backend doesn't support recursion or your measurements show a benefit. A compact recursive grammar doesn't automatically suffer state-space explosion. A flat parent_id representation also introduces cross-record constraints that syntax can't enforce:

from dataclasses import dataclass

@dataclass(frozen=True)
class FlatNode:
    node_id: str
    parent_id: str | None
    label: str

nodes = [
    FlatNode(node_id="root", parent_id=None, label="ci_run"),
    FlatNode(node_id="n1", parent_id="root", label="failing_job"),
    FlatNode(node_id="n2", parent_id="root", label="log_url"),
]

def validate_tree(nodes: list[FlatNode], max_nodes=100):
    if not 1 <= len(nodes) <= max_nodes:
        raise ValueError("node count outside policy")
    by_id = {node.node_id: node for node in nodes}
    if len(by_id) != len(nodes) or any(not node.node_id.strip() for node in nodes):
        raise ValueError("node IDs must be unique and nonempty")
    roots = [node.node_id for node in nodes if node.parent_id is None]
    if len(roots) != 1:
        raise ValueError("expected one root")
    children = {node.node_id: [] for node in nodes}
    for node in nodes:
        if node.parent_id is not None:
            if node.parent_id not in by_id:
                raise ValueError("unknown parent")
            children[node.parent_id].append(node.node_id)
    pending, visited = [roots[0]], set()
    while pending:
        current = pending.pop()
        if current in visited:
            raise ValueError("cycle detected")
        visited.add(current)
        pending.extend(children[current])
    if len(visited) != len(nodes):
        raise ValueError("disconnected nodes or cycles present")
    return roots[0], children

root_id, children = validate_tree(nodes)
print("root ID:", root_id)
print("child IDs:", children[root_id])

cyclic = [nodes[0], FlatNode("a", "b", "job"), FlatNode("b", "a", "log")]
try:
    validate_tree(cyclic)
except ValueError as error:
    print("invalid tree:", error)

root ID: root
child IDs: ['n1', 'n2']
invalid tree: disconnected nodes or cycles present

The fixed-depth wire shape avoids recursive JSON nesting, but the host must now enforce unique IDs, existing parents, one root, and connectedness. This fixture checks those relationships. Flattening trades one set of constraints for another; it doesn't make arbitrary graph integrity a regular-language property.

Critical failure modes and resilient production cascades

Even with grammar-guided decoding, production deployments encounter distinct failure modes that require defensive handling.

An incomplete prefix can still be legal

Grammar legality doesn't guarantee progress toward termination. A JSON grammar that allows arbitrary whitespace can keep accepting spaces before an object. A broad string rule can accept more characters without closing its quote.

Low model scores for legal tokens aren't a grammatical dead end: finite legal logits can still be renormalized. A true dead end has no representable legal continuation, for example because of an incompatible grammar/tokenizer or a backend error.

Possible failures include:

• Emitting endless repeated whitespace (\n \n ) if the grammar permits optional indentation.
• Repeating characters where the grammar permits them. A valid JSON array can't contain an endless sequence of commas without intervening values.
• Generating repetitive nonsensical strings that satisfy a broad regex until hitting max_output_tokens.

A grammar constrains syntax, not task understanding or termination. Clear task instructions can help; they don't guarantee recovery. Inspect the grammar, tokenizer compatibility, stop rules, and incomplete-output telemetry, and retain a finite token/time budget.

The "no prompting needed" fallacy

A common misconception is that strict grammar enforcement eliminates the need for detailed prompt engineering.

The mask filters continuations; task instructions explain what the fields mean and how to represent missing evidence. Some providers inject format instructions automatically, so pasting the entire schema into every prompt isn't universally necessary. Specify the extraction task, source boundaries, and unknown-value policy, then evaluate whether field descriptions or examples improve the result.

Version schemas in application code

Schemas evolve over time. If producers and consumers disagree on field definitions during a rollout, services can crash even when outputs are syntactically valid JSON:

def validate_run_status_v2(payload: object) -> dict[str, str | None]:
    return RunStatusUpdate.model_validate(payload).model_dump()

generated_payload = {"run_id": "RUN-842", "status": "failed", "failing_job": "unit-tests"}
validated = validate_run_status_v2(generated_payload)
wire_record = {"schema_version": "2", "payload": validated}

print("version:", wire_record["schema_version"])
print("failing job:", wire_record["payload"]["failing_job"])

version: 2
failing job: unit-tests

Application code must own the schema version. Don't let the model generate the version identifier, because a model's claim about which schema it followed is untrusted.

The production fallback cascade

Choose recovery by failure class, rather than downgrading every failure through one cascade:

1. Completed candidate: Parse and validate it, then check source facts and authorization. Pydantic only checks domain rules you actually implement; it doesn't supply RBAC or truth checks automatically.
2. Token limit: A bounded retry with the same schema and more tokens may help if it fits the model limit and your cost/deadline policy. It doesn't guarantee success. OpenAI Chat Completions uses finish_reason: "length"; Responses uses status: "incomplete" with incomplete_details.reason: "max_output_tokens". Don't mix these wire contracts.^{[16]Reference 16Structured outputshttps://developers.openai.com/api/docs/guides/structured-outputs}
3. Refusal or content-filter interruption: Stop automatic retries and use your application's safe refusal path. Don't loosen validation or safety instructions to obtain an answer.
4. Unsupported/invalid schema: Fix the request configuration. Repeating the same invalid schema won't compile it successfully.
5. Local validation failure: Diagnose a contract mismatch or unsupported constraint. A correction retry can be appropriate for a recoverable data error, with a bounded attempt count and a sanitized diagnostic. Never send unrelated secrets or raw authorization failures back to the model as a repair prompt.

For a provider without constrained generation, treat prompt-only output as an untrusted candidate. Heuristic JSON repair may change meaning; schema acceptance after repair doesn't establish fidelity to the original response. An explicit reject/re-request path is easier to audit than silently inventing missing values.

The following deterministic dictionary-response fixture exercises recovery decisions. It isn't a drop-in OpenAI SDK client: the SDK returns typed response objects, and real requests also need a supported model and transport-error handling. The budgets 120 and 400 are illustrative limits, not a mandated doubling rule.

FORMAT = {"type": "json_schema", "name": "run_status", "strict": True, "schema": schema}

class FixtureClient:
    def __init__(self, responses):
        self.responses = iter(responses)
        self.calls = []

    def create(self, **kwargs):
        self.calls.append(kwargs)
        return next(self.responses)

def generate_with_bounded_retry(client, input_text: str) -> RunStatusUpdate:
    for attempt, budget in enumerate((120, 400)):
        response = client.create(input=input_text, text={"format": FORMAT}, max_output_tokens=budget)
        output = response.get("output", [])
        content = [part for item in output if item.get("type") == "message"
                   for part in item.get("content", [])]
        if any(part.get("type") == "refusal" for part in content):
            raise RuntimeError("policy refusal: request rejected by safety filter")
        status = response.get("status")
        if status == "incomplete":
            reason = (response.get("incomplete_details") or {}).get("reason")
            if reason == "max_output_tokens" and attempt == 0:
                continue
            raise RuntimeError(f"halted prematurely: {reason}")
        if status != "completed":
            raise RuntimeError(f"unexpected completion status: {status}")
        messages = [item for item in output if item.get("type") == "message"]
        if len(messages) != 1 or messages[0].get("status") != "completed":
            raise RuntimeError("expected one completed output message")
        if len(content) != 1 or content[0].get("type") != "output_text":
            raise RuntimeError("expected one structured text item")
        return RunStatusUpdate.model_validate(load_record(content[0]["text"]))
    raise RuntimeError("retry budget exhausted")

incomplete = {"status": "incomplete", "output": [],
              "incomplete_details": {"reason": "max_output_tokens"}}
completed = {"status": "completed", "output": [{"type": "message", "status": "completed",
             "content": [{"type": "output_text", "text": json.dumps(generated_payload)}]}]}

client = FixtureClient([incomplete, completed])
update = generate_with_bounded_retry(client, SOURCE_LOG)

print("retry preserved contract:", update.model_dump())
print("budgets:", [call["max_output_tokens"] for call in client.calls])
assert client.calls[0]["text"] == client.calls[1]["text"]

refused = {"status": "completed", "output": [{"type": "reasoning", "summary": []},
           {"type": "message", "status": "completed", "content": [{"type": "refusal", "refusal": "blocked"}]}]}
blocked_client = FixtureClient([refused])

try:
    generate_with_bounded_retry(blocked_client, SOURCE_LOG)
except RuntimeError as error:
    print("refusal routed:", error)

assert len(blocked_client.calls) == 1

filtered = {"status": "incomplete", "output": [],
            "incomplete_details": {"reason": "content_filter"}}
for label, responses in (("content filter", [filtered]),
                         ("second length stop", [incomplete, incomplete])):
    stopped_client = FixtureClient(responses)
    try:
        generate_with_bounded_retry(stopped_client, SOURCE_LOG)
    except RuntimeError as error:
        print(f"{label}: {error}; calls={len(stopped_client.calls)}")
    assert len(stopped_client.calls) == len(responses)

retry preserved contract: {'run_id': 'RUN-842', 'status': 'failed', 'failing_job': 'unit-tests'}
budgets: [120, 400]
refusal routed: policy refusal: request rejected by safety filter
content filter: halted prematurely: content_filter; calls=1
second length stop: halted prematurely: max_output_tokens; calls=2

This fixture permits one length retry with an expanded budget under the exact same schema. A second length stop, a refusal, or a content-filter interruption terminates recovery. In an application, check cost and deadline policy before granting the retry; don't commit a partial record while waiting.

A schema field is not an authorization grant

A dangerous failure mode in structured LLM interfaces is schema-valid false authority: boolean or enum fields that look like authorization decisions.

For example, a model might emit {"is_admin": true, "authorized": true, "run_id": "RUN-842"}. The payload matches your JSON Schema, but treating model output as an authorization grant introduces a severe privilege escalation vulnerability:

from dataclasses import dataclass

@dataclass(frozen=True)
class ModelRunProposal:
    run_id: str
    action: str
    approved: bool  # Never treat model-generated flags as authorization grants

def approval_gate(proposal: ModelRunProposal, stored_request: tuple[str, str] | None) -> str:
    if proposal.action != "retry_tests":
        return "deny: action outside policy"
    if stored_request != (proposal.run_id, proposal.action):
        return "deny: missing approval for this request"
    return "approval matched; continue host checks"

proposal = ModelRunProposal("RUN-842", "retry_tests", approved=True)
print(approval_gate(proposal, stored_request=None))
print(approval_gate(proposal, stored_request=("RUN-999", "retry_tests")))
print(approval_gate(proposal, stored_request=("RUN-842", "retry_tests")))
print("model approved field was", proposal.approved, "and was ignored")

deny: missing approval for this request
deny: missing approval for this request
approval matched; continue host checks
model approved field was True and was ignored

Model outputs are proposals, not permissions. Authorization grants must come from trusted host session tokens, database permissions, and verified user identity.

Practice: build an incident-digest parser

Now apply these principles by implementing an incident digest extraction tool.

You receive an unstructured 500-word incident report describing engineering outages. Your service must extract all mentioned components and classify their operational state as healthy, degraded, outage, or unknown.

System requirements

1. Return a mentions array with one current-state summary per service. Each item contains service (string) and status (enum). Multiple mentions in the source must be reconciled from their timing and evidence, not array position.
2. Enforce the schema using structured outputs or grammar-guided decoding.
3. Represent the absence of service mentions as an empty list {"mentions": []}, not null or an omitted key.
4. Implement a post-parsing semantic validation gate that rejects duplicate service summaries and cross-references services against a known service catalog (auth-api, vector-indexer, billing-export), flagging unrecognized services for operator review.

Sample input log

"auth-api recovered after elevated 5xx errors. vector-indexer remains degraded. report-worker has no known impact."

Target JSON structure

{
  "mentions": [
    {"service": "auth-api", "status": "healthy"},
    {"service": "vector-indexer", "status": "degraded"},
    {"service": "report-worker", "status": "unknown"}
  ]
}

Reference validator

This validator checks supplied candidates; it doesn't implement model-based extraction. Add the constrained-generation call yourself, then test omitted mentions, unsupported status claims, unknown catalog entries, and an empty report. A correctly shaped record can still miss a service or misread its state.

ALLOWED_STATUS = {"healthy", "degraded", "outage", "unknown"}
KNOWN_SERVICES = {"auth-api", "vector-indexer", "billing-export"}

def validate_digest(payload: object) -> dict[str, list[dict[str, str]]]:
    if not isinstance(payload, dict) or set(payload) != {"mentions"}:
        raise ValueError("root must be an object with only mentions")
    mentions = payload["mentions"]
    if not isinstance(mentions, list):
        raise TypeError("mentions must be a list")
    checked: list[dict[str, str]] = []
    for item in mentions:
        if not isinstance(item, dict):
            raise TypeError("each mention must be an object")
        extra = sorted(set(item) - {"service", "status"})
        missing = [key for key in ("service", "status") if key not in item]
        if extra or missing:
            raise ValueError(f"mention shape extra={extra} missing={missing}")
        service = item["service"]
        status = item["status"]
        if (
            not isinstance(service, str)
            or not service.strip()
            or not isinstance(status, str)
            or status not in ALLOWED_STATUS
        ):
            raise TypeError("service must be a string and status must be a known enum")
        checked.append({"service": service, "status": status})
    return {"mentions": checked}

def unknown_services(digest: dict[str, list[dict[str, str]]]) -> list[str]:
    return [
        mention["service"]
        for mention in digest["mentions"]
        if mention["service"] not in KNOWN_SERVICES
    ]

def check_service_identity(digest: dict[str, list[dict[str, str]]]) -> None:
    seen = set()
    for mention in digest["mentions"]:
        service = mention["service"]
        if service in seen:
            raise ValueError(f"duplicate service summary: {service}")
        seen.add(service)

digest = validate_digest(
    {
        "mentions": [
            {"service": "auth-api", "status": "healthy"},
            {"service": "vector-indexer", "status": "degraded"},
            {"service": "report-worker", "status": "unknown"},
        ]
    }
)
empty = validate_digest({"mentions": []})
check_service_identity(digest)

print("mentions:", len(digest["mentions"]))
print("unknown services:", unknown_services(digest))
print("empty mentions:", empty["mentions"])

conflicting = validate_digest({"mentions": [
    {"service": "auth-api", "status": "healthy"},
    {"service": "auth-api", "status": "outage"},
]})
last_wins = {item["service"]: item["status"] for item in conflicting["mentions"]}
print("tempting dict conversion:", last_wins)
try:
    check_service_identity(conflicting)
except ValueError as error:
    print("identity gate:", error)

mentions: 3
unknown services: ['report-worker']
empty mentions: []
tempting dict conversion: {'auth-api': 'outage'}
identity gate: duplicate service summary: auth-api

Where structured outputs fit in production systems

Structured output generation gives probabilistic models a constrained data interface. Its practical contract depends on the provider, supported schema subset, compiled grammar, and completion state. A typed candidate is still a candidate.

Enforcing structure doesn't eliminate the need for application-level safety gates:

• Grammar-guided decoding enforces syntax and schema structure during generation.
• Pydantic models enforce client-side type invariants and custom field boundaries.
• Application services enforce tenant isolation, RBAC scopes, and ground-truth validation.
• Resilient fallback cascades manage truncation retries, safety refusals, and targeted reflection.

Before connecting a record to a side effect, answer three questions: did the response complete under the intended contract, does current evidence support the proposal, and does trusted application policy permit this caller to perform that operation?