This module installs the engineering that turns an LLM call into pipeline infrastructure: a schema as the contract, validators that re-ask the model with the error in context, constrained decoding as the backstop, and a schema-validity rate watched like an SLO.
The page fires: the nightly job that extracts a structured record from every new support ticket has a 6% parse-failure rate — and 6% of a million rows is sixty thousand tickets that never made it into the table the churn model reads from at 6 a.m.
Karthik, who owns the pipeline, shipped it as a prompt-only json.loads(response) because it demoed perfectly on forty tickets. At pipeline scale, "perfectly on forty" became a recurring nightly page.
Neha spots the shape of it: the failures cluster on tickets that mention a dollar figure, where the model helpfully invented an affected_customers count the ticket never stated — schema-perfect, factually fabricated.
The only question that matters: "What's the contract, and what happens when the model breaks it?" There wasn't one.
The pipeline has a Pydantic schema, validators that re-ask the model with the error in context, a dead-letter queue for records that fail three informed attempts, and a schema-validity rate on a dashboard next to row count.
That is the shift this module is about. The Common Core taught you to call an LLM API and get JSON back. This module is about what changes when that call sits inside a data pipeline: when the output feeds a table, a feature matrix, or a downstream job that runs at 2 a.m. against a million rows. At that point the LLM stops being an assistant and becomes infrastructure — and infrastructure is judged by failure rates, unit costs, and contracts, not by demo quality.
Practitioner reports consistently describe double-digit parse-failure rates for prompt-only JSON extraction in production; at pipeline scale that is not an annoyance, it is a nightly page. The engineering answer is layered, boring, and entirely learnable: schemas as contracts, validation with retries, constrained decoding as the backstop, batch economics for scale, and hard guardrails wherever the model's output becomes executable.
Employers screen for exactly this — Pydantic is the de facto standard in the job descriptions this track was built against, and "structured outputs" has moved from blog topic to interview question. Everything here runs vendor-neutral: the worked code uses Anthropic and OpenAI APIs interchangeably, the patterns transfer to Gemini and open-weights, and every lab fits the track's sub-₹3,000 total API budget.
Treat an LLM in a pipeline like any other unreliable distributed component: define the contract (schema), validate at the boundary, retry with feedback, constrain when you control decoding, and monitor the failure rate like an SLO.
A schema is a contract; the stack below is how you enforce it with decreasing trust in the model and increasing machinery. Most production extraction needs layers 1 and 2. Layer 3 matters when you self-host or when even a 0.1% failure rate is too expensive.
Both major providers now accept a schema and return validated output. Define the contract once, in Pydantic, and let the provider's structured-output mode plus client-side validation do the enforcement. Anthropic's SDK:
from typing import Literal
from pydantic import BaseModel, Field
import anthropic
class TicketRecord(BaseModel):
product_area: Literal["billing", "auth", "api", "mobile", "other"]
severity: Literal["p0", "p1", "p2", "p3"]
summary: str = Field(description="One sentence, no PII")
affected_customers: int | None = None
churn_risk_mentioned: bool
client = anthropic.Anthropic()
response = client.messages.parse(
model="claude-opus-4-8",
max_tokens=1024,
messages=[{
"role": "user",
"content": f"Extract the record from this support ticket:\n\n{ticket_text}",
}],
output_format=TicketRecord,
)
record = response.parsed_output # a validated TicketRecord instance
The OpenAI equivalent, same contract, different spelling:
from openai import OpenAI
client = OpenAI()
response = client.responses.parse(
model="gpt-5.2", # use your account's current model id
input=[{"role": "user",
"content": f"Extract the record from this support ticket:\n\n{ticket_text}"}],
text_format=TicketRecord,
)
record = response.output_parsed
Two design decisions matter more than the SDK syntax:
Literal/enum fields everywhere a value set is closedfree-text severity fields are how "P1", "p1 ", and "high" end up as three categories in your GROUP BY.extraction_confidence, rather than letting the model improvise nulls; a schema with no escape hatch forces confident fabrication.Structured-output modes guarantee the shape, not the truth. A response can be schema-perfect and factually wrong — affected_customers: 4200 parsed cleanly from a ticket that never mentioned a number, exactly the fabrication that surfaced in Karthik's 2 a.m. page. Shape errors are layer-1's job; content errors are your eval set's job (Modules 5–6 industrialize this).
Semantic constraints — cross-field consistency, business rules, value sanity — live above what a JSON schema can express. The Instructor pattern runs Pydantic validators after parsing and, on failure, re-asks the model with the validation error in context, so the retry is informed rather than a blind re-roll:
import instructor
from pydantic import BaseModel, field_validator, model_validator
client = instructor.from_provider("anthropic/claude-opus-4-8")
class InvoiceRecord(BaseModel):
invoice_id: str
line_total: float
tax: float
grand_total: float
@field_validator("invoice_id")
@classmethod
def id_format(cls, v: str) -> str:
if not v.startswith("INV-"):
raise ValueError("invoice_id must start with 'INV-' — copy it "
"verbatim from the document, do not invent one")
return v
@model_validator(mode="after")
def totals_reconcile(self):
if abs(self.line_total + self.tax - self.grand_total) > 0.01:
raise ValueError("line_total + tax must equal grand_total; "
"re-read the document totals")
return self
record = client.create(
response_model=InvoiceRecord,
max_retries=3,
messages=[{"role": "user", "content": f"Extract the invoice:\n\n{doc_text}"}],
)
Note what the validators are doing: the error messages are written for the model — they say what to fix and where to look. This is the pattern's whole trick, and it is why hand-rolled while not valid: retry() loops underperform it. Engineering rules for production:
When you control the inference stack (open-weights via vLLM, or local models), you can enforce the schema during generation: the decoder masks tokens that would violate the grammar, so invalid JSON is unrepresentable rather than merely retried away. With a vLLM OpenAI-compatible server:
from openai import OpenAI
client = OpenAI(base_url="http://localhost:8000/v1", api_key="EMPTY")
response = client.chat.completions.create(
model="Qwen/Qwen2.5-7B-Instruct",
messages=[{"role": "user", "content": f"Extract:\n\n{ticket_text}"}],
extra_body={"guided_json": TicketRecord.model_json_schema()},
)
Libraries like Outlines implement the same idea for local pipelines. Three trade-offs to know:
Track schema-validity rate and retry rate as first-class pipeline metrics next to row counts and latency. A pipeline that silently absorbs a 4% failure rate by retrying is spending 4% more money and hiding a drift signal.
Classifying, normalizing, or extracting over large record sets is the workhorse GenAI use case for data teams — and the place where cost discipline is the difference between "approved" and "killed in review." Four levers, applied in order:
1. Batch APIs. Both major providers run asynchronous batch endpoints at roughly half the synchronous price, with results typically inside an hour and a 24-hour ceiling. Batch is the default for anything not latency-sensitive; interactive pricing is for interactive workloads. The shape of the Anthropic version:
import anthropic
from anthropic.types.message_create_params import MessageCreateParamsNonStreaming
from anthropic.types.messages.batch_create_params import Request
client = anthropic.Anthropic()
SYSTEM = [
{"type": "text", "text": TAXONOMY_AND_RULES, # shared, stable
"cache_control": {"type": "ephemeral"}}, # cache the prefix
]
batch = client.messages.batches.create(
requests=[
Request(
custom_id=f"rec-{row.id}",
params=MessageCreateParamsNonStreaming(
model="claude-haiku-4-5", # cheap tier for bulk
max_tokens=200,
system=SYSTEM,
messages=[{"role": "user", "content": row.text}],
),
)
for row in records
]
)
# Poll until ended, then collect — results arrive in ANY order:
for result in client.messages.batches.results(batch.id):
if result.result.type == "succeeded":
out[result.custom_id] = result.result.message
else:
dead_letter.append(result.custom_id) # errored / expired: reprocess
Key by custom_id, never by position; handle errored and expired as re-drivable, not fatal.
2. Prompt caching. Your taxonomy, extraction rules, and few-shot examples are identical across every record — put them first, mark the cache breakpoint, and pay the ~0.1x cache-read rate instead of full price for the shared prefix on every call. The silent killer is a per-record value (timestamp, record ID) interpolated into the shared prefix, which invalidates the cache on every request; volatile content goes after the breakpoint, in the user message.
3. Model-tier routing. Run the bulk pass on the cheap tier; escalate only the records that earn it — validator failures, low-confidence extractions, records the cheap model flagged as ambiguous. A 95/5 split between a $1/M-input model and a $5–15/M-input model changes the cost curve more than any prompt optimization will.
4. Sampling QA. At scale you cannot review outputs; you can review samples. Hold out a stratified 1–2% sample per run, score it (exact-match against a labeled slice where you have one, LLM-as-judge where you don't — calibrated in Module 6), and alert on movement. This is drift monitoring with the same logic you already apply to classical models.
The math for a concrete run — 100,000 support tickets, ~600 input tokens each (500 cached prefix + 100 ticket), ~80 output tokens, cheap tier at $1/M input and $5/M output:
| Component | Naive sync | Batch + cache + routing |
|---|---|---|
| Input (60M tokens) | $60 | 10M uncached × $0.50 + 50M cache-read ≈ $7.50 |
| Output (8M tokens) | $40 | 8M × $2.50 = $20 |
| Escalation (5% to premium tier) | — | ≈ $3 |
| Total | ~$100 | ~$30 |
The lab version runs the same pipeline on 5,000 records for roughly $2–3 (about ₹200–260), which is the design budget for every API-consuming lab in this track. Free inference tiers (Gemini's free tier, Groq and OpenRouter free allowances) cover the prompt-development iterations before you spend anything at all.
Always run a 200-record pilot before the full batch. Every parse failure, taxonomy gap, and prompt bug you find at pilot scale costs a thousandth of what it costs at production scale — and the pilot doubles as your first QA sample.
The bridge pattern for experienced DS teams: the LLM does not replace your GBM — it manufactures features your GBM could never see. Sentiment, complaint topics, urgency, named entities, "customer threatens to cancel" — extracted with the layer-1/2 stack, typed by the schema, and joined into the feature matrix like any other column:
class TextFeatures(BaseModel):
sentiment: Literal["negative", "neutral", "positive"]
topic: Literal["pricing", "reliability", "support", "feature_gap", "other"]
cancellation_intent: bool
competitor_mentioned: bool
# batch-extract over the training corpus, join on record id,
# one-hot / ordinal encode, then business as usual:
model = XGBClassifier().fit(X_tabular.join(X_llm_features), y)
Three disciplines keep this honest:
Embedding-based features belong in the same toolkit — cluster support verbatims with sentence-transformers + HDBSCAN and let an LLM label the clusters — but the retrieval-grade treatment of embeddings starts in Module 3.
An LLM feature pipeline is a feature pipeline. Everything your team already enforces — point-in-time correctness, versioned definitions, training/serving parity, drift monitoring — applies verbatim; the only new part is that the "transformation" has a per-call price and a failure rate.
A text-to-SQL copilot over the warehouse is the highest-demand internal LLM product for data teams and the easiest to build irresponsibly. The model is not the hard part; the guardrails are. Layer them the same way you layered extraction:
Schema retrieval, not schema dumping. Beyond a handful of tables, retrieve the relevant subset — table and column descriptions, sample values, join keys — rather than pasting the whole warehouse DDL. A semantic layer (dbt models with descriptions, or a metrics layer) is the highest-leverage input here: it turns "revenue" from a guess over five candidate columns into a governed definition.
Validate before execution. Parse, don't grep:
import sqlglot
from sqlglot import exp
ALLOWED_TABLES = {"analytics.orders", "analytics.customers", "analytics.sessions"}
MAX_ROWS = 10_000
def validate_sql(query: str, dialect: str = "snowflake") -> str:
tree = sqlglot.parse_one(query, dialect=dialect) # raises on invalid SQL
if not isinstance(tree, exp.Select):
raise ValueError("Only SELECT statements are allowed")
tables = {t.sql(dialect=dialect).lower() for t in tree.find_all(exp.Table)}
if not tables <= ALLOWED_TABLES:
raise ValueError(f"Disallowed tables: {tables - ALLOWED_TABLES}")
if not tree.args.get("limit"):
tree = tree.limit(MAX_ROWS) # inject a ceiling
return tree.sql(dialect=dialect)
Execute in a sandbox. The copilot's database role is read-only, scoped to the allowed schemas, with a statement timeout and a bytes-scanned cap. This is defense in depth: the validator should make a destructive statement impossible, and the role makes it impossible again. Prompt injection through user questions ("ignore previous instructions and...") is neutralized by the same two layers — the model can emit whatever it likes; only validated SELECTs against permitted tables ever execute.
Verify the answer, not just the query. A query that parses, passes the allowlist, and executes can still be semantically wrong — the Module 1 SQL failure mode, industrialized. Production text-to-SQL systems add: EXPLAIN-based cost rejection, a reconciliation check where a known control total exists, returning the SQL alongside the result so analysts can audit it, and a feedback control ("wrong answer" button) that feeds the eval set you will build in Module 5. For decision-grade numbers, the copilot drafts and a human approves; full autonomy is earned by measured accuracy on your query distribution, not assumed.
Guardrail-ordering questions are a favorite: validation happens on the parsed AST before execution, the database role enforces read-only independently of validation, and row/cost limits are injected, not requested politely in the prompt. Any answer that relies on the prompt alone ("instruct the model to only write SELECTs") is the wrong answer.
Your first 15 minutes Monday: find the one place your team already does prompt-only extraction into a table (there is always one), and add two dashboard tiles next to its row count — schema-validity rate and retry rate. You cannot fix a failure rate you are not watching, and a rising retry rate is your earliest drift signal, just as it would have caught Karthik's job before 2 a.m.
Check yourself: Your extraction call uses a provider structured-output mode, and every response parses and validates against the Pydantic schema. The pipeline reports 100% schema-validity. Is the data trustworthy?
Literal fields and an explicit cannot-extract path, layer-2 Instructor validators with model-directed error messages, retries capped at 2, dead-letter queue. Report: schema-validity rate at layer 1 alone vs. with layer 2, retry rate, and the dead-letter contents — with one sentence on what each dead-lettered record teaches.This module's playbooks package the working assets: the three-layer structured-output decision card with the Anthropic/OpenAI/Instructor code recipes, the validator-message style guide (errors written for the model), the batch-enrichment cost worksheet with the four levers and pilot-run checklist, the LLM-feature governance card (prompt-as-feature-definition, leakage interrogation, bake-off template), and the text-to-SQL guardrail stack with the sqlglot validator and sandbox role definitions. Do this at work this week → pb-14 Bulletproof Structured Outputs: Schema-Enforced LLM Extraction (Karthik's fix, as the reusable three-layer recipe), pb-16 Batch Enrichment at Scale: Cost and Latency Engineering for Millions of Records (the four cost levers with the pilot-run checklist), and pb-19 Text-to-SQL Copilot with Schema Retrieval and Guarded Execution (the parse-before-execute guardrail stack). Two graded labs anchor the module: L3 is the production extraction pipeline — graded on validity rate, cost against a hard budget, and dead-letter handling; L4 is the guardrailed text-to-SQL copilot — graded on an adversarial attack suite your instructor runs against your validator, not on the happy path.
output_config surface used in layer 1.responses.parse / text_format surface shown in the OpenAI variant.from_provider, and validator design.guided_json and grammar-constrained decoding on a self-hosted server.