Structured outputs — three ways

Track 04 - Agents · Notebook 02 · Runtime: ~5s on CPU

Prerequisites: 04_agents/01 (ReAct from scratch).

Three strategies for getting a model to emit a parseable schema:

1. Prompt-only — ask, parse, accept failures.

2. Validate + retry — parse, on ValidationError send the error back.

3. FSM-constrained (Outlines, Willard & Louf 2307.09702) — modify sampling so only schema-conforming tokens are generated.

To compare them with controlled failure rates, we use a deterministic test-bench whose outputs distribute as ~70% valid, ~15% schema-violating, ~15% malformed JSON — the failure modes real LLMs actually emit.

For the head-to-head against a real model (Anthropic tool-use, Instructor, prompt-only, validate+retry on the same 200 prompts), see 08_production/04_structured_outputs_real.

from llm_systems_cookbook.nb import bootstrap

import json
import re

import numpy as np
from pydantic import BaseModel, Field, ValidationError

s = bootstrap("04_agents_02_structured_outputs_three_ways")

The schema

We want a Person object with a name, age in [0, 130], and a list of hobbies. Pydantic gives us both validation and a JSON schema for the FSM-constrained path.

class Person(BaseModel):
    name: str = Field(min_length=1, max_length=50)
    age: int = Field(ge=0, le=130)
    hobbies: list[str] = Field(default_factory=list, max_length=10)


SCHEMA = Person.model_json_schema()
print("JSON schema top-level keys:", list(SCHEMA.keys()))
print(f"required fields: {SCHEMA.get('required', [])}")

Controlled-failure test-bench

A FlakyOutputs source produces three categories of string output, in proportions that match what real models emit at the loose end of prompt quality:

• Valid + compliant (70%): JSON that parses and validates.

• Valid but violating (15%): JSON that parses but fails the Pydantic schema (out-of-range age, etc.).

• Malformed (15%): pre-/post-amble text, trailing commas.

The point is to expose each strategy to all three failure classes deterministically, so we can compare compliance and call-count.

rng = np.random.default_rng(0)

GOOD_NAMES = ["Alice", "Bob", "Carol", "Dan", "Eve"]
GOOD_HOBBIES = [["chess"], ["hiking", "painting"], [], ["cooking"]]


class FlakyOutputs:
    """Deterministic test-bench. Not an LLM — a controlled-failure source."""

    def __init__(self) -> None:
        self.rng = np.random.default_rng(0)

    def generate(self, prompt: str, retry_correction: str | None = None) -> str:
        # On retry, always produce valid output (a real model uses the error
        # message effectively most of the time).
        if retry_correction is not None:
            name = self.rng.choice(GOOD_NAMES)
            age = int(self.rng.integers(10, 80))
            hobbies = GOOD_HOBBIES[int(self.rng.integers(0, len(GOOD_HOBBIES)))]
            return json.dumps({"name": str(name), "age": age, "hobbies": list(hobbies)})

        r = self.rng.random()
        if r < 0.70:
            name = self.rng.choice(GOOD_NAMES)
            age = int(self.rng.integers(10, 80))
            hobbies = GOOD_HOBBIES[int(self.rng.integers(0, len(GOOD_HOBBIES)))]
            return json.dumps({"name": str(name), "age": age, "hobbies": list(hobbies)})
        if r < 0.85:
            # Valid JSON, schema-violating.
            name = self.rng.choice(GOOD_NAMES)
            bad_age = int(self.rng.choice([-5, 200, 999]))
            return json.dumps({"name": str(name), "age": bad_age, "hobbies": []})
        # Malformed JSON: pre-amble, trailing comma.
        name = self.rng.choice(GOOD_NAMES)
        return f"Sure! Here is the object: {{\"name\": \"{name}\", \"age\": 42,}}"


def extract_first_json(text: str) -> str | None:
    start = text.find("{")
    end = text.rfind("}")
    if start == -1 or end == -1 or end <= start:
        return None
    return text[start : end + 1]

Three strategies

Each takes an LLM instance and a prompt and returns either a Person or None.

def strategy_prompt_only(src: FlakyOutputs, prompt: str) -> Person | None:
    raw = src.generate(prompt)
    chunk = extract_first_json(raw)
    if chunk is None:
        return None
    try:
        data = json.loads(chunk)
        return Person(**data)
    except (json.JSONDecodeError, ValidationError, TypeError):
        return None


def strategy_validate_retry(src: FlakyOutputs, prompt: str, max_retries: int = 2) -> Person | None:
    correction: str | None = None
    for _ in range(max_retries + 1):
        raw = src.generate(prompt, retry_correction=correction)
        chunk = extract_first_json(raw) or raw
        try:
            data = json.loads(chunk)
        except json.JSONDecodeError as e:
            correction = f"Your output failed JSON parsing: {e.msg}. Output ONLY valid JSON."
            continue
        try:
            return Person(**data)
        except (ValidationError, TypeError) as e:
            correction = f"Your output failed schema validation: {e}. Fix the fields and re-emit."
            continue
    return None


def strategy_fsm_constrained(src: FlakyOutputs, prompt: str) -> Person:
    """Stand-in for FSM-constrained decoding. In production, Outlines (or the
    provider's native `tool_use`) restricts the sampler to only emit tokens
    that satisfy the JSON schema. We model the *outcome* — always-valid
    output — without re-implementing the FSM. See `08_production/04` for the
    real-model comparison."""
    name = rng.choice(GOOD_NAMES)
    age = int(rng.integers(10, 80))
    hobbies = GOOD_HOBBIES[int(rng.integers(0, len(GOOD_HOBBIES)))]
    return Person(name=str(name), age=age, hobbies=list(hobbies))

Run and score

200 trials per strategy. Metrics:

• validity - what fraction return a non-None Person.

• latency cost - number of LLM calls made (always 1 for prompt-only and FSM, up to 3 for validate+retry).

N_TRIALS = 200

def run(strategy, N: int = N_TRIALS) -> tuple[float, int]:
    src = FlakyOutputs()
    valid = 0
    total_calls = 0
    for _ in range(N):
        out = strategy(src, "Generate a Person object")
        if out is not None and isinstance(out, Person):
            valid += 1
        total_calls += 1
    return valid / N, total_calls


valid_po, _  = run(strategy_prompt_only)
valid_vr, _  = run(strategy_validate_retry)
valid_fsm, _ = run(strategy_fsm_constrained)

print(f"prompt-only        validity = {valid_po:.3f}")
print(f"validate+retry     validity = {valid_vr:.3f}")
print(f"FSM-constrained    validity = {valid_fsm:.3f}")

s.check(
    "prompt_only_matches_flaky_baseline",
    lambda: 0.55 <= valid_po <= 0.80,
    msg=f"validity = {valid_po:.3f}  (expected ~0.70)",
)
s.check(
    "validate_retry_lifts_validity_over_prompt_only",
    lambda: valid_vr > valid_po,
    msg=f"prompt-only={valid_po:.3f}  validate+retry={valid_vr:.3f}",
)
s.check(
    "fsm_constrained_is_perfect",
    lambda: valid_fsm == 1.0,
    msg=f"validity = {valid_fsm:.3f}",
)
s.check(
    "retry_validity_near_one",
    lambda: valid_vr >= 0.98,
    msg=f"validate+retry validity = {valid_vr:.3f}",
)
# Pydantic rejects out-of-range ages.
bad_ages_rejected = True
for a in (-1, 131, 1000):
    try:
        Person(name="x", age=a, hobbies=[])
        bad_ages_rejected = False
    except ValidationError:
        pass
s.check("pydantic_enforces_age_range", lambda: bad_ages_rejected)

Validity vs latency - the three-way tradeoff

Two bar charts side by side: schema compliance on the left, average LLM calls per successful output on the right. Prompt-only is cheapest but leaks malformed outputs at the base rate of the model. Validate+retry climbs to near-perfect compliance but pays extra calls on failures. FSM-constrained is perfect at exactly one call per output, which is why it has become the default in production function-calling APIs.

import matplotlib.pyplot as plt


class CountingFlaky(FlakyOutputs):
    def __init__(self):
        super().__init__(); self.calls = 0
    def generate(self, prompt, retry_correction=None):
        self.calls += 1; return super().generate(prompt, retry_correction)


def measure(strategy, N=N_TRIALS):
    src = CountingFlaky()
    ok = sum(1 for _ in range(N) if strategy(src, "Generate a Person object") is not None)
    return ok / N, src.calls / N


po  = measure(strategy_prompt_only)
vr  = measure(strategy_validate_retry)
fsm = (valid_fsm, 1.0)

names = ["prompt-only", "validate+retry", "FSM-constrained"]
validity = [po[0], vr[0], fsm[0]]
calls = [po[1], vr[1], fsm[1]]

fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(9, 3.2))
colors = ["tab:orange", "tab:blue", "tab:green"]
ax1.bar(names, validity, color=colors)
ax1.set_ylim(0, 1.05); ax1.set_ylabel("validity (fraction)")
ax1.set_title("schema compliance across 200 trials")
for i, v in enumerate(validity):
    ax1.text(i, v + 0.02, f"{v:.2f}", ha="center")

ax2.bar(names, calls, color=colors)
ax2.set_ylabel("avg calls per output")
ax2.set_title("call cost (calls ≈ wall time)")
for i, v in enumerate(calls):
    ax2.text(i, v + 0.02, f"{v:.2f}", ha="center")
fig.tight_layout(); plt.show()

Exercises

1. Real Outlines. pip install outlines and run outlines.generate.json(model, Person) against a local Qwen2.5 on the same 200 prompts. Compare the call count and latency.

2. Cost-vs-quality curve. Sweep max_retries ∈ {0, 1, 2, 3, 5} for validate+retry; plot validity vs total calls per successful output.

3. Nested schema. Add an Address submodel. Re-run prompt-only against a real LLM (use 08_production/04_structured_outputs_real as the harness) — nested JSON is where prompt-only falls off.

References

• Willard & Louf 2023, Efficient Guided Generation for LLMs — the FSM-decoding algorithm.

• Pydantic v2 docs on model_validate_json and TypeAdapter.validate_python.

• OpenAI / Anthropic / Google function-calling docs — strategy 3 pushed into the inference layer.

• 08_production/04_structured_outputs_real — the same comparison against a real model.

s.summary()
s.save()