Structured outputs — four real strategies head-to-head

Open in Colab Run this notebook in Colab

Structured outputs — four real strategies head-to-head#

Track 08 - Production · Notebook 04 · Runtime: ~5 min LIVE, <1s replay

Same Pydantic schema, same 200 prompts, same model — different schema-enforcement strategies. We measure schema compliance, cost, and the extra calls each retry strategy buys.

from llm_systems_cookbook.nb import bootstrap
from llm_systems_cookbook._utils import repo_root

import json
import os

from pydantic import BaseModel, Field, ValidationError

s = bootstrap("08_production_04_structured_outputs_real")

LIVE = bool(os.environ.get("ANTHROPIC_API_KEY"))
MODEL = os.environ.get("MODEL_ANTHROPIC", "claude-haiku-4-5")
FIXTURE = json.loads((repo_root() / "notebooks/08_production/_fixtures/04_structured.json").read_text())
print(f"mode={'LIVE' if LIVE else 'REPLAY'}  model={MODEL}  N={FIXTURE['n_prompts']}")

Schema#

A small Person schema with two interesting failure modes: out-of-range age (Pydantic catches with Field(ge=0, le=130)) and a hobbies list with bounded length.

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.dumps({"required": SCHEMA["required"], "props": list(SCHEMA["properties"])}))

Four strategies, one Anthropic SDK#

All four call the same model; only the schema-enforcement layer changes.

# 1. Prompt-only — ask for JSON, parse it. The optimist.
def strategy_prompt_only(prompt: str) -> Person | None:
    if not LIVE:
        # In replay we sample from the recorded compliance distribution.
        return None  # actual code path is below; replay uses aggregates

    import anthropic  # noqa: PLC0415
    client = anthropic.Anthropic()
    resp = client.messages.create(
        model=MODEL, max_tokens=200,
        system="Reply with ONLY a JSON object matching {name, age, hobbies}.",
        messages=[{"role": "user", "content": prompt}],
    )
    text = "".join(b.text for b in resp.content if b.type == "text")
    try:
        start, end = text.find("{"), text.rfind("}")
        return Person.model_validate_json(text[start : end + 1])
    except (ValidationError, ValueError, json.JSONDecodeError):
        return None

# 2. Validate + retry — parse, on ValidationError feed the error back.
def strategy_validate_retry(prompt: str, max_retries: int = 2) -> Person | None:
    if not LIVE: return None  # see replay note

    import anthropic  # noqa: PLC0415
    client = anthropic.Anthropic()
    messages = [{"role": "user", "content": prompt}]
    system = "Reply with ONLY a JSON object matching {name, age, hobbies}."
    for _ in range(max_retries + 1):
        resp = client.messages.create(model=MODEL, max_tokens=200,
                                       system=system, messages=messages)
        text = "".join(b.text for b in resp.content if b.type == "text")
        try:
            start, end = text.find("{"), text.rfind("}")
            return Person.model_validate_json(text[start : end + 1])
        except (ValidationError, ValueError) as e:
            messages += [
                {"role": "assistant", "content": text},
                {"role": "user", "content": f"That failed validation: {e}. Re-emit valid JSON."},
            ]
    return None

# 3. Tool-use — schema is a tool, model must call it.
PERSON_TOOL = {
    "name": "record_person",
    "description": "Record a person extracted from the user message.",
    "input_schema": SCHEMA,
}

def strategy_tool_use(prompt: str) -> Person | None:
    if not LIVE: return None

    import anthropic  # noqa: PLC0415
    client = anthropic.Anthropic()
    resp = client.messages.create(
        model=MODEL, max_tokens=300,
        tools=[PERSON_TOOL],
        tool_choice={"type": "tool", "name": "record_person"},
        messages=[{"role": "user", "content": prompt}],
    )
    for block in resp.content:
        if block.type == "tool_use" and block.name == "record_person":
            try:
                return Person.model_validate(block.input)
            except ValidationError:
                return None
    return None

# 4. Instructor — wraps tool_use with a typed return + automatic retry.
def strategy_instructor(prompt: str, max_retries: int = 2) -> Person | None:
    if not LIVE: return None

    import anthropic, instructor  # noqa: PLC0415
    client = instructor.from_anthropic(anthropic.Anthropic())
    return client.messages.create(
        model=MODEL, max_tokens=300,
        response_model=Person,
        max_retries=max_retries,
        messages=[{"role": "user", "content": prompt}],
    )

STRATEGIES = {
    "prompt_only":     strategy_prompt_only,
    "validate_retry":  strategy_validate_retry,
    "tool_use":        strategy_tool_use,
    "instructor":      strategy_instructor,
}
print(f"strategies: {list(STRATEGIES)}")

Aggregate over 200 prompts#

Live mode runs every strategy on the full prompt set (~5 min, ~$0.40 in API calls). Replay loads the aggregates from a recorded run; the distributions are real.

def aggregate_live(strategy_fn, name: str, prompts: list[str]) -> dict:
    """LIVE-mode aggregator — returns the same shape as the fixture entries."""
    import time as _time  # noqa: PLC0415
    successes, total_calls, total_cost_estim, total_lat = 0, 0, 0.0, 0.0
    for p in prompts:
        t0 = _time.perf_counter()
        out = strategy_fn(p)
        total_lat += _time.perf_counter() - t0
        # Crude estimate; real notebook would read response.usage.
        total_calls += 1
        total_cost_estim += 0.0005
        if out is not None: successes += 1
    return {"compliance": successes / len(prompts),
            "mean_latency_s": total_lat / len(prompts),
            "mean_cost_usd": total_cost_estim / len(prompts),
            "mean_calls": total_calls / max(successes, 1)}


if LIVE:
    PROMPTS = [
        "Alice is 31 and likes chess and hiking.",
        # ...in practice you would load 200 prompts here
    ]
    summary = {n: aggregate_live(fn, n, PROMPTS) for n, fn in STRATEGIES.items()}
else:
    summary = FIXTURE["results"]

print(f"{'strategy':<16} {'compliance':>10} {'lat (s)':>9} {'$/call':>10} {'calls/ok':>10}")
for name, r in summary.items():
    print(f"{name:<16} {r['compliance']:>10.1%} {r['mean_latency_s']:>9.2f} "
          f"${r['mean_cost_usd']:>8.6f} {r['mean_calls']:>10.2f}")

Cost vs compliance Pareto#

import matplotlib.pyplot as plt

names = list(summary)
compl = [summary[n]["compliance"] * 100 for n in names]
costs = [summary[n]["mean_cost_usd"] * 1000 for n in names]
calls = [summary[n]["mean_calls"] for n in names]

fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(10, 3.4))
colors = ["tab:red", "tab:orange", "tab:blue", "tab:green"]

ax1.scatter(costs, compl, s=120, c=colors)
for i, n in enumerate(names):
    ax1.annotate(n, (costs[i], compl[i]), xytext=(6, -3),
                 textcoords="offset points", fontsize=9)
ax1.set_xlabel("mean cost per call (×1e-3 $)")
ax1.set_ylabel("schema compliance (%)")
ax1.set_title(f"compliance vs cost · {FIXTURE['n_prompts']} prompts")
ax1.set_ylim(80, 102); ax1.grid(alpha=0.3)

ax2.bar(names, calls, color=colors)
ax2.set_ylabel("LLM calls per successful output")
ax2.set_title("retry overhead")
ax2.axhline(1.0, color="black", linewidth=0.8, alpha=0.4)
for i, v in enumerate(calls):
    ax2.text(i, v + 0.005, f"{v:.2f}", ha="center", fontsize=9)

fig.tight_layout(); plt.show()

Checks#

s.check(
    "tool_use_perfect_compliance",
    lambda: summary["tool_use"]["compliance"] == 1.0,
    msg=f"tool_use compliance = {summary['tool_use']['compliance']:.3f}",
)
s.check(
    "instructor_perfect_compliance",
    lambda: summary["instructor"]["compliance"] == 1.0,
    msg=f"instructor compliance = {summary['instructor']['compliance']:.3f}",
)
s.check(
    "validate_retry_lifts_above_prompt_only",
    lambda: summary["validate_retry"]["compliance"] > summary["prompt_only"]["compliance"] + 0.05,
    msg=(f"prompt_only={summary['prompt_only']['compliance']:.3f} "
         f"retry={summary['validate_retry']['compliance']:.3f}"),
)
s.check(
    "tool_use_avoids_extra_calls",
    lambda: summary["tool_use"]["mean_calls"] <= 1.01,
    msg=f"tool_use mean calls/ok = {summary['tool_use']['mean_calls']:.3f}",
)
# Pydantic enforces age range — a quick local check that the schema is real.
def _bad_age_rejected():
    try:
        Person(name="x", age=150, hobbies=[]); return False
    except ValidationError:
        return True
s.check("pydantic_rejects_out_of_range_age", _bad_age_rejected)

Notes for production#

  • Default to tool-use. When the provider supports it (Anthropic, OpenAI, Gemini, Mistral) tool-use beats validate+retry on every axis: always-valid output, no retry budget, no failure recovery code.

  • Instructor adds DX, not capability. It generates the tool spec from your Pydantic model, retries on ValidationError, and types the return. Same wire calls underneath.

  • Outlines for local models. When you serve your own model (Qwen2.5, Llama 3.x via vLLM), Outlines compiles the schema to a finite-state machine and constrains token-level sampling. 100% compliance with no retries even on a 1.5B model. Provider APIs do this internally; you can only get it locally.

  • BAML and TypeChat are alternatives in the same niche — BAML compiles a separate schema language; TypeChat uses TypeScript types. All converge on tool-use under the hood.

  • Structured outputs ≠ structured reasoning. A perfectly-validated Person.age = 200 is still wrong. Validation only catches shape; semantic validity needs prompts, examples, or a downstream check.

s.summary()
s.save()