Imagine you've built a document extraction agent that pulls structured data from insurance claims. It works beautifully in your notebook. You ship it to production, and three days later, your downstream pipeline is on fire because the model returned "claim_amount": "not specified" instead of a number, or decided to wrap its JSON in a markdown code block, or worse—hallucinated a field called clam_amount that silently passed through your deserialization layer and corrupted your database. Nobody caught it. The agent was "working."
This is the fundamental problem with AI agent outputs: language models are probabilistic systems, and probabilistic systems do not honor contracts by default. You have to enforce those contracts. In high-stakes production environments—financial data extraction, medical record processing, legal document analysis—an agent that returns plausible-looking but structurally invalid output is not a feature. It's a liability. The gap between "the model usually returns the right shape" and "the model is contractually bound to return the right shape" is where production AI systems live or die.
By the end of this lesson, you'll know how to build that enforcement layer from first principles. We'll cover JSON Schema design for LLM outputs, OpenAI's structured outputs API and its limitations, Pydantic-based validation pipelines, retry logic with error context injection, and how to assemble these pieces into a production-grade validation architecture. You'll also understand the subtle failure modes that catch experienced engineers off guard.
What you'll learn:
response_format with strict: true and understand its constraints at the grammar level$ref, required, and additionalProperties mean)Before we write a single line of validation code, we need to understand why LLM outputs fail structurally—because the failure modes dictate your defensive strategy.
Language models generate text token by token, sampling from a probability distribution at each step. When you ask a model to return JSON, it isn't running a JSON serializer. It's generating characters that look like JSON based on training data patterns. This means several failure categories exist simultaneously:
Syntactic failures are the obvious ones: unmatched braces, trailing commas (valid in JavaScript, not JSON), single quotes instead of double quotes, or a model that decides to explain its answer before or after the JSON blob. These are easy to detect but surprisingly common in long-context tasks where the model "forgets" it was supposed to be in JSON mode.
Semantic failures are more dangerous: the JSON is valid, it even deserializes without error, but the values are wrong types ("1234.56" instead of 1234.56), required fields are missing or null when they shouldn't be, enum values are slightly wrong ("IN_PROGRESS" instead of "in_progress"), or numeric values are out of acceptable ranges.
Hallucination failures are the sneakiest: the model adds extra fields that aren't in your schema ("confidence_score", "note", "reasoning"), and if your deserialization doesn't check for unexpected keys, these fields silently get dropped or, worse, accepted.
There's also a fourth category that only becomes apparent at scale: distribution shift failures. Your agent works fine for 99.3% of inputs, but on certain document types—say, insurance claims from a specific regional carrier that uses unusual terminology—the model consistently misclassifies a field or drops a nested object. You won't catch this with unit tests. You need runtime validation with alerting.
The implication is that your defense strategy needs to be layered: constrain what the model can output at the generation level, validate what it does output at the parsing level, and repair or retry when validation fails. Let's build each layer.
The first line of defense is constraining generation itself. Modern LLM APIs offer mechanisms to bias or force the model toward valid structured outputs. The most powerful of these, as of 2024–2025, is OpenAI's structured outputs feature.
OpenAI's response_format: { type: "json_schema", json_schema: {...}, strict: true } works by converting your JSON Schema into a context-free grammar and then using that grammar to constrain the token sampling process. At each token generation step, only tokens that could legally continue a valid document according to the grammar are permitted. This means the model cannot produce syntactically invalid JSON or JSON that violates the schema structure.
This is a fundamentally different guarantee than "json_mode," which only ensures syntactically valid JSON. With strict: true, you're guaranteed that the output will conform to the structural constraints in your schema. But there are important limitations:
Not all JSON Schema features are supported. OpenAI's grammar engine supports a specific subset: object, array, string, number, integer, boolean, null, enum, anyOf, $ref, $defs. It does not support pattern, minimum/maximum, minLength/maxLength, format, or many other constraint keywords. Those validations have to happen in your post-processing layer.
additionalProperties must be false. This is required for strict: true. This is actually a feature—it forces you to enumerate every field—but it's a breaking change if you're used to flexible schemas.
All fields must be explicitly required or wrapped in anyOf with null. The model cannot omit fields; it must explicitly return null for optional ones.
Let's look at what a well-designed structured output schema looks like for our insurance claim extractor:
from openai import OpenAI
import json
client = OpenAI()
CLAIM_SCHEMA = {
"name": "insurance_claim_extraction",
"strict": True,
"schema": {
"type": "object",
"properties": {
"claim_id": {
"type": "string",
"description": "The unique claim identifier, typically formatted as CLM-XXXXXXXX"
},
"claimant_name": {
"type": "string",
"description": "Full legal name of the person filing the claim"
},
"claim_amount": {
"anyOf": [
{"type": "number"},
{"type": "null"}
],
"description": "Total claimed amount in USD. Null if not specified in the document."
},
"claim_date": {
"type": "string",
"description": "Date the claim was filed, in YYYY-MM-DD format"
},
"claim_status": {
"type": "string",
"enum": ["pending", "approved", "denied", "under_review", "closed"],
"description": "Current processing status of the claim"
},
"line_items": {
"type": "array",
"items": {
"$ref": "#/$defs/line_item"
}
},
"notes": {
"anyOf": [
{"type": "string"},
{"type": "null"}
],
"description": "Any additional notes or flags from the document. Null if none."
}
},
"required": [
"claim_id", "claimant_name", "claim_amount",
"claim_date", "claim_status", "line_items", "notes"
],
"additionalProperties": False,
"$defs": {
"line_item": {
"type": "object",
"properties": {
"description": {"type": "string"},
"amount": {"type": "number"},
"category": {
"type": "string",
"enum": ["medical", "property", "liability", "legal", "other"]
}
},
"required": ["description", "amount", "category"],
"additionalProperties": False
}
}
}
}
def extract_claim(document_text: str) -> dict:
response = client.chat.completions.create(
model="gpt-4o-2024-08-06", # Structured outputs require this model or later
messages=[
{
"role": "system",
"content": (
"You are an expert insurance document analyst. Extract the structured "
"claim data from the provided document. If a field is not present in "
"the document, return null. For claim_date, always convert to YYYY-MM-DD format."
)
},
{
"role": "user",
"content": f"Extract the claim data from this document:\n\n{document_text}"
}
],
response_format={
"type": "json_schema",
"json_schema": CLAIM_SCHEMA
}
)
return json.loads(response.choices[0].message.content)
Notice several design decisions here. We use anyOf with null for fields that genuinely might not exist in a document. We provide detailed description fields—these are read by the model and significantly improve extraction quality. We use $defs to define reusable sub-schemas rather than inlining them, which keeps the schema maintainable and is required for recursive schemas.
Warning: Don't confuse
strict: truein the JSON schema object with validation strictness. Thestrictflag controls whether OpenAI enforces the schema at generation time. It does not validate semantic constraints like number ranges, string patterns, or cross-field relationships. You still need post-generation validation.
When using structured outputs, the model can still refuse to process content that violates OpenAI's policies. In this case, the message.content will be None and message.refusal will contain the refusal message. Always check for this:
def extract_claim_safe(document_text: str) -> dict | None:
response = client.chat.completions.create(
model="gpt-4o-2024-08-06",
messages=[...],
response_format={"type": "json_schema", "json_schema": CLAIM_SCHEMA}
)
message = response.choices[0].message
if message.refusal:
# Log this — refusals in document extraction are usually a content policy
# false positive on sensitive medical or legal content
raise ValueError(f"Model refused to process document: {message.refusal}")
return json.loads(message.content)
Grammar-constrained generation ensures structural validity. But your business logic has requirements that grammars can't express. Pydantic v2 is where you enforce those.
The critical insight here is that your Pydantic models should be additive on top of your JSON Schema—they enforce constraints that the schema can't express. Don't just reimplement the schema in Pydantic; that's redundant work. Add the business logic layer:
from pydantic import BaseModel, Field, field_validator, model_validator
from datetime import date, datetime
from decimal import Decimal
from typing import Literal
import re
class LineItem(BaseModel):
description: str = Field(min_length=1, max_length=500)
amount: float = Field(ge=0.0) # Can't have negative line items
category: Literal["medical", "property", "liability", "legal", "other"]
@field_validator("amount")
@classmethod
def amount_must_be_reasonable(cls, v: float) -> float:
# Individual line items over $10M are almost certainly extraction errors
if v > 10_000_000:
raise ValueError(
f"Line item amount {v} exceeds maximum reasonable value of $10,000,000. "
f"This is likely a parsing error (e.g., the model included cents as a separate amount)."
)
return round(v, 2)
class InsuranceClaim(BaseModel):
claim_id: str
claimant_name: str = Field(min_length=2, max_length=200)
claim_amount: float | None
claim_date: str # We'll parse and validate this
claim_status: Literal["pending", "approved", "denied", "under_review", "closed"]
line_items: list[LineItem]
notes: str | None
# Parsed date — not in the LLM output, computed during validation
parsed_claim_date: date | None = Field(default=None, exclude=True)
@field_validator("claim_id")
@classmethod
def validate_claim_id_format(cls, v: str) -> str:
# Our system uses CLM- prefix followed by 8 alphanumeric characters
if not re.match(r'^CLM-[A-Z0-9]{8}$', v):
raise ValueError(
f"claim_id '{v}' does not match expected format CLM-XXXXXXXX. "
f"The model may have extracted an internal reference number instead of the claim ID."
)
return v
@field_validator("claim_date")
@classmethod
def validate_and_parse_date(cls, v: str) -> str:
try:
parsed = datetime.strptime(v, "%Y-%m-%d").date()
except ValueError:
# Try to give the model a hint about what went wrong
raise ValueError(
f"claim_date '{v}' is not in YYYY-MM-DD format. "
f"Common errors: MM/DD/YYYY, DD-MM-YYYY, or month names like 'January 15, 2024'."
)
# Claims can't be dated in the future
if parsed > date.today():
raise ValueError(
f"claim_date {v} is in the future ({parsed}). "
f"This is likely a transcription error or the wrong date field was extracted."
)
# Claims older than 10 years are suspicious for active processing
if (date.today() - parsed).days > 3650:
raise ValueError(
f"claim_date {v} is more than 10 years ago. "
f"Verify this is the filing date and not a policy inception date."
)
return v
@model_validator(mode="after")
def validate_amount_consistency(self) -> "InsuranceClaim":
"""Cross-field validation: claim_amount should roughly match sum of line items."""
if self.line_items and self.claim_amount is not None:
line_item_total = sum(item.amount for item in self.line_items)
if line_item_total > 0 and self.claim_amount > 0:
discrepancy_ratio = abs(self.claim_amount - line_item_total) / line_item_total
# If claim amount differs from line item total by more than 20%, flag it
if discrepancy_ratio > 0.20:
raise ValueError(
f"claim_amount ({self.claim_amount}) differs from sum of line items "
f"({line_item_total:.2f}) by {discrepancy_ratio:.1%}. "
f"The model may have extracted a subtotal or pre-deductible amount. "
f"Verify which total figure to use."
)
return self
The error messages here are written specifically to be useful when injected back into the model as retry context. We'll use them in the next layer. Notice how each error message describes not just what failed but why the model might have made this mistake and what the correct behavior should be. This is deliberate—these strings will become part of a correction prompt.
Tip: The
exclude=Truefield parameter onparsed_claim_datemeans it won't be serialized when you callmodel.model_dump(). This is useful for computed fields that are internal to your validation layer but shouldn't flow downstream.
Here's where most production implementations fall short. When validation fails, the naive approach is to either crash or silently return None. The sophisticated approach is to retry with the validation error injected into the model's context. This works remarkably well because Pydantic error messages—when written carefully—provide exactly the kind of specific, actionable feedback that models respond to.
import json
import logging
from typing import TypeVar, Type, Callable, Any
from pydantic import BaseModel, ValidationError
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
logger = logging.getLogger(__name__)
T = TypeVar("T", bound=BaseModel)
class ExtractionError(Exception):
"""Raised when extraction fails after all retries are exhausted."""
def __init__(self, message: str, attempts: list[dict], last_raw_output: str | None):
super().__init__(message)
self.attempts = attempts
self.last_raw_output = last_raw_output
class ValidationRetryError(Exception):
"""Raised to trigger a retry due to validation failure. Not a final error."""
pass
def build_correction_prompt(
original_output: str,
validation_errors: list[dict]
) -> str:
"""
Build a structured correction prompt from Pydantic validation errors.
This prompt will be injected as a user message in the retry conversation.
"""
error_descriptions = []
for error in validation_errors:
location = " -> ".join(str(loc) for loc in error["loc"]) if error["loc"] else "root"
error_descriptions.append(
f" Field: {location}\n"
f" Error: {error['msg']}\n"
f" Your value: {error.get('input', 'not provided')}"
)
errors_formatted = "\n\n".join(error_descriptions)
return (
f"Your previous extraction attempt had the following validation errors:\n\n"
f"{errors_formatted}\n\n"
f"Your previous output was:\n```json\n{original_output}\n```\n\n"
f"Please correct these specific issues and return the complete, corrected JSON. "
f"Do not change fields that were correct."
)
def extract_with_retry(
document_text: str,
model_class: Type[T],
schema: dict,
max_attempts: int = 3,
system_prompt: str = "",
) -> T:
"""
Extract structured data from text with automatic retry on validation failure.
Each failed attempt injects the validation errors back into the conversation
as a correction request. The model sees its previous output and specific
error descriptions.
"""
conversation_history = [
{
"role": "system",
"content": system_prompt or "Extract structured data from the provided document."
},
{
"role": "user",
"content": f"Extract the data from this document:\n\n{document_text}"
}
]
attempts_log = []
last_raw_output = None
for attempt_num in range(1, max_attempts + 1):
logger.info(f"Extraction attempt {attempt_num}/{max_attempts}")
try:
response = client.chat.completions.create(
model="gpt-4o-2024-08-06",
messages=conversation_history,
response_format={
"type": "json_schema",
"json_schema": schema
}
)
message = response.choices[0].message
if message.refusal:
raise ExtractionError(
f"Model refused: {message.refusal}",
attempts=attempts_log,
last_raw_output=None
)
raw_output = message.content
last_raw_output = raw_output
raw_data = json.loads(raw_output)
# Attempt Pydantic validation
validated = model_class.model_validate(raw_data)
# Success — log the attempt and return
attempts_log.append({
"attempt": attempt_num,
"status": "success",
"raw_output": raw_output
})
if attempt_num > 1:
logger.info(f"Extraction succeeded on attempt {attempt_num}")
return validated
except ValidationError as e:
error_details = e.errors(include_url=False)
attempts_log.append({
"attempt": attempt_num,
"status": "validation_failure",
"raw_output": last_raw_output,
"errors": error_details
})
logger.warning(
f"Attempt {attempt_num} failed validation with {len(error_details)} errors: "
f"{[err['msg'] for err in error_details]}"
)
if attempt_num == max_attempts:
raise ExtractionError(
f"Extraction failed after {max_attempts} attempts. "
f"Last errors: {[err['msg'] for err in error_details]}",
attempts=attempts_log,
last_raw_output=last_raw_output
)
# Add the model's previous response to history
conversation_history.append({
"role": "assistant",
"content": last_raw_output
})
# Inject correction prompt
correction_prompt = build_correction_prompt(last_raw_output, error_details)
conversation_history.append({
"role": "user",
"content": correction_prompt
})
except json.JSONDecodeError as e:
# This shouldn't happen with structured outputs, but handle it anyway
attempts_log.append({
"attempt": attempt_num,
"status": "json_parse_failure",
"raw_output": last_raw_output,
"error": str(e)
})
if attempt_num == max_attempts:
raise ExtractionError(
f"JSON parsing failed after {max_attempts} attempts",
attempts=attempts_log,
last_raw_output=last_raw_output
)
The conversation-based retry is more effective than simply re-running the extraction because the model can see what it produced and why it was wrong. This is the difference between a 60% self-correction rate and a 90%+ self-correction rate on common validation errors.
Warning: Watch your token counts in retry loops. Each retry appends both the previous response and the correction prompt to the conversation history. For long documents with large extracted objects, this can push you toward context window limits on retry 3 or 4. Consider truncating the document in the correction prompt if you're working with large inputs.
Individual extraction with retry is one thing. A production pipeline processes hundreds or thousands of documents concurrently, needs observability, and needs graceful degradation when things go wrong systematically. Let's build the orchestration layer.
import asyncio
from openai import AsyncOpenAI
from dataclasses import dataclass, field
from datetime import datetime
from collections import defaultdict
import time
async_client = AsyncOpenAI()
@dataclass
class ExtractionResult:
document_id: str
success: bool
data: InsuranceClaim | None
error: str | None
attempts: int
processing_time_ms: float
timestamp: datetime = field(default_factory=datetime.utcnow)
@dataclass
class PipelineMetrics:
total: int = 0
succeeded: int = 0
failed: int = 0
retried: int = 0
total_attempts: int = 0
error_types: dict = field(default_factory=lambda: defaultdict(int))
avg_processing_time_ms: float = 0.0
class ExtractionPipeline:
def __init__(
self,
max_concurrency: int = 10,
max_retries: int = 3,
circuit_breaker_threshold: float = 0.5,
circuit_breaker_window: int = 20
):
self.semaphore = asyncio.Semaphore(max_concurrency)
self.max_retries = max_retries
self.metrics = PipelineMetrics()
self._recent_results: list[bool] = []
self._circuit_open = False
self._circuit_breaker_threshold = circuit_breaker_threshold
self._circuit_breaker_window = circuit_breaker_window
def _update_circuit_breaker(self, success: bool):
"""
Simple sliding window circuit breaker.
If failure rate in the last N requests exceeds threshold, open the circuit.
"""
self._recent_results.append(success)
if len(self._recent_results) > self._circuit_breaker_window:
self._recent_results.pop(0)
if len(self._recent_results) >= self._circuit_breaker_window:
failure_rate = self._recent_results.count(False) / len(self._recent_results)
if failure_rate > self._circuit_breaker_threshold and not self._circuit_open:
logger.error(
f"Circuit breaker OPENED: {failure_rate:.1%} failure rate over "
f"last {self._circuit_breaker_window} requests. "
f"This may indicate a schema change, model degradation, or upstream data issue."
)
self._circuit_open = True
elif failure_rate <= self._circuit_breaker_threshold / 2 and self._circuit_open:
logger.info("Circuit breaker CLOSED: failure rate has recovered")
self._circuit_open = False
async def extract_single_async(
self,
document_id: str,
document_text: str,
) -> ExtractionResult:
start_time = time.monotonic()
async with self.semaphore:
if self._circuit_open:
return ExtractionResult(
document_id=document_id,
success=False,
data=None,
error="Circuit breaker open — pipeline is experiencing high failure rate",
attempts=0,
processing_time_ms=0
)
conversation_history = [
{
"role": "system",
"content": (
"You are an expert insurance document analyst. Extract structured "
"claim data. Use null for any field not present in the document. "
"For dates, always use YYYY-MM-DD format."
)
},
{
"role": "user",
"content": f"Extract claim data:\n\n{document_text}"
}
]
last_raw_output = None
for attempt in range(1, self.max_retries + 1):
try:
response = await async_client.chat.completions.create(
model="gpt-4o-2024-08-06",
messages=conversation_history,
response_format={
"type": "json_schema",
"json_schema": CLAIM_SCHEMA
}
)
raw_output = response.choices[0].message.content
last_raw_output = raw_output
data = InsuranceClaim.model_validate(json.loads(raw_output))
elapsed = (time.monotonic() - start_time) * 1000
self._update_circuit_breaker(True)
return ExtractionResult(
document_id=document_id,
success=True,
data=data,
error=None,
attempts=attempt,
processing_time_ms=elapsed
)
except ValidationError as e:
if attempt < self.max_retries:
conversation_history.append(
{"role": "assistant", "content": last_raw_output}
)
conversation_history.append({
"role": "user",
"content": build_correction_prompt(
last_raw_output,
e.errors(include_url=False)
)
})
else:
elapsed = (time.monotonic() - start_time) * 1000
self._update_circuit_breaker(False)
return ExtractionResult(
document_id=document_id,
success=False,
data=None,
error=f"Validation failed: {e.error_count()} errors after {attempt} attempts",
attempts=attempt,
processing_time_ms=elapsed
)
except Exception as e:
elapsed = (time.monotonic() - start_time) * 1000
self._update_circuit_breaker(False)
return ExtractionResult(
document_id=document_id,
success=False,
data=None,
error=f"Unexpected error: {type(e).__name__}: {str(e)}",
attempts=attempt,
processing_time_ms=elapsed
)
async def process_batch(
self,
documents: list[tuple[str, str]] # (document_id, document_text)
) -> list[ExtractionResult]:
tasks = [
self.extract_single_async(doc_id, doc_text)
for doc_id, doc_text in documents
]
results = await asyncio.gather(*tasks, return_exceptions=False)
# Update aggregate metrics
for result in results:
self.metrics.total += 1
self.metrics.total_attempts += result.attempts
if result.success:
self.metrics.succeeded += 1
else:
self.metrics.failed += 1
if result.attempts > 1:
self.metrics.retried += 1
success_count = sum(1 for r in results if r.success)
logger.info(
f"Batch complete: {success_count}/{len(results)} succeeded, "
f"{sum(r.attempts for r in results)} total API calls"
)
return results
async def main():
pipeline = ExtractionPipeline(
max_concurrency=15,
max_retries=3,
circuit_breaker_threshold=0.4,
circuit_breaker_window=30
)
# documents loaded from your storage layer
documents = [
("CLM-20240001", "INSURANCE CLAIM FORM\nClaim ID: CLM-AB123456\n..."),
("CLM-20240002", "..."),
# ... hundreds more
]
results = await pipeline.process_batch(documents)
# Separate successes from failures
successful = [r for r in results if r.success]
failed = [r for r in results if not r.success]
# Route failed extractions to human review queue
for failure in failed:
await route_to_human_review(failure.document_id, failure.error)
# Process successful extractions
claims = [r.data for r in successful]
await persist_claims(claims)
print(f"Pipeline metrics: {pipeline.metrics}")
asyncio.run(main())
In production, your extraction schemas will change over time. New fields get added, enum values expand, validation rules tighten. You need a strategy for managing this without breaking existing data or pipelines.
from enum import Enum
from typing import ClassVar
import hashlib
class SchemaVersion(str, Enum):
V1_0 = "1.0"
V1_1 = "1.1"
V2_0 = "2.0"
class VersionedInsuranceClaim(InsuranceClaim):
schema_version: str = Field(default=SchemaVersion.V2_0, exclude=False)
# Class-level schema fingerprint for detecting silent schema drift
SCHEMA_FINGERPRINT: ClassVar[str] = ""
@classmethod
def compute_schema_fingerprint(cls) -> str:
"""Generate a hash of the current schema for change detection."""
schema_json = json.dumps(cls.model_json_schema(), sort_keys=True)
return hashlib.sha256(schema_json.encode()).hexdigest()[:12]
# Register schema fingerprint at startup
VersionedInsuranceClaim.SCHEMA_FINGERPRINT = (
VersionedInsuranceClaim.compute_schema_fingerprint()
)
This fingerprint approach gives you a way to detect when your Pydantic model schema has drifted from what's stored in your database or what your JSON Schema sends to the model. Include it in your health check endpoint.
When you release a schema update, you'll have documents extracted with the old schema living in your database. A migration validator handles this:
def migrate_claim_v1_to_v2(raw_data: dict) -> dict:
"""
Migrate a v1.0 claim extraction to v2.0 schema.
v2.0 added the 'line_items' field and made 'notes' nullable.
"""
migrated = raw_data.copy()
# v1 didn't have line_items — initialize as empty
if "line_items" not in migrated:
migrated["line_items"] = []
# v1 used "status" instead of "claim_status"
if "status" in migrated and "claim_status" not in migrated:
migrated["claim_status"] = migrated.pop("status")
# v1 notes field was sometimes a list of strings
if isinstance(migrated.get("notes"), list):
migrated["notes"] = "; ".join(migrated["notes"]) if migrated["notes"] else None
migrated["schema_version"] = SchemaVersion.V2_0
return migrated
def load_claim_with_migration(raw_data: dict) -> VersionedInsuranceClaim:
version = raw_data.get("schema_version", "1.0")
if version == "1.0":
raw_data = migrate_claim_v1_to_v2(raw_data)
return VersionedInsuranceClaim.model_validate(raw_data)
Validation pipelines generate rich signal that most teams ignore. Here's what to capture and why.
import structlog
log = structlog.get_logger()
def log_extraction_event(result: ExtractionResult, document_metadata: dict):
log_data = {
"event": "claim_extraction",
"document_id": result.document_id,
"success": result.success,
"attempts": result.attempts,
"processing_time_ms": result.processing_time_ms,
"retried": result.attempts > 1,
**document_metadata
}
if result.success:
log_data.update({
"claim_status": result.data.claim_status,
"has_line_items": len(result.data.line_items) > 0,
"line_item_count": len(result.data.line_items),
"claim_amount_present": result.data.claim_amount is not None,
})
log.info("extraction_success", **log_data)
else:
log_data["error"] = result.error
log.warning("extraction_failure", **log_data)
The key metrics to track in your monitoring dashboard are:
claim_date is failing 8% of the time, there's likely a date format your prompt isn't handling.You'll build a complete extraction pipeline for financial earnings reports. This is a more complex extraction task than insurance claims because financial documents have heterogeneous structure and values that require cross-validation.
Scenario: You're building an automated financial data pipeline that extracts key metrics from public company earnings reports. The pipeline must extract revenue, net income, EPS, and guidance data while ensuring the extracted numbers are internally consistent.
Part 1: Schema Design
Design a JSON Schema for the following data structure:
EarningsReport:
- company_ticker: string (1-5 uppercase letters)
- fiscal_quarter: string (format: "Q1 2024", "Q2 2024", etc.)
- revenue_millions: number | null
- net_income_millions: number | null
- earnings_per_share: number | null
- revenue_guidance_low: number | null
- revenue_guidance_high: number | null
- key_metrics: array of {metric_name: string, value: number, unit: string}
- management_tone: enum of ["positive", "neutral", "cautious", "negative"]
Requirements: strict: true, all optional fields wrapped in anyOf with null.
Part 2: Pydantic Model with Business Validators
Add these validations:
company_ticker must match ^[A-Z]{1,5}$revenue_guidance_low must be ≤ revenue_guidance_high when both are presentPart 3: Retry Logic
Implement extract_earnings_with_retry() using the conversation injection pattern. Test it by deliberately providing a malformed extraction (wrong date format, inconsistent EPS sign) and verify the correction prompt causes the model to fix it.
Part 4: Observability
Add logging that captures:
management_tone values (useful for sentiment analysis)response_format: { type: "json_object" } (json_mode) only guarantees syntactically valid JSON. It does not enforce any schema. Yet many teams use it and are surprised when the model returns wrong field names or types. Always use json_schema with strict: true when you have a known schema.
A Pydantic error message like "Value error, invalid claim ID" is useless in a correction prompt. The model doesn't know what a valid claim ID looks like. Always include: what the value was, what format was expected, and why the model commonly makes this mistake. Compare:
# Bad
raise ValueError("Invalid date format")
# Good
raise ValueError(
f"'{v}' is not a valid date. Expected YYYY-MM-DD (e.g., '2024-03-15'). "
f"Common mistake: extracting MM/DD/YYYY from American-format documents."
)
If your validation logic has a bug that makes a field permanently fail, an unbounded retry loop will spin up infinite API calls. Always have a hard limit on attempts and ensure your ExtractionError contains full context for debugging.
Some teams deserialize to a plain dict, pass it through their pipeline, and validate "at the end." This means invalid data can corrupt intermediate state. Always validate at the boundary—the moment you receive data from the model.
When the model hits the max_tokens limit mid-JSON, finish_reason will be "length" and the content will be truncated, invalid JSON (or the structured output system will return an error). Always check finish_reason == "stop" before attempting to parse:
choice = response.choices[0]
if choice.finish_reason != "stop":
raise ValueError(
f"Model stopped with reason '{choice.finish_reason}'. "
f"Output may be truncated. Consider increasing max_tokens or reducing document length."
)
When you tighten a validation rule (e.g., adding a regex constraint to claim_id), run it against your historical extracted data before deploying. New validators frequently fail on edge cases in real data that your test suite never covered. Maintain a regression test dataset of real production examples.
If retry rates spike for a subset of documents, the diagnostic path is:
You've built a complete structured output contract system with four interlocking layers: grammar-constrained generation with json_schema and strict: true, semantic validation with Pydantic business logic, intelligent retry with error context injection, and async pipeline orchestration with circuit breaking and observability.
The key architectural insight is that each layer catches different failure modes, and no single layer is sufficient on its own. Grammar constraints can't validate business rules. Pydantic can't constrain token generation. Retry logic is ineffective without informative error messages. Observability is useless without the right metrics. Together, they create a system where probabilistic model outputs are forced to meet deterministic contracts.
What to build next:
Structured output testing framework: Build a test suite that runs your schema and Pydantic models against a corpus of real documents (including adversarial edge cases) before every schema change. This is the highest-leverage investment for long-term reliability.
Human-in-the-loop fallback: For documents that fail all retries, build a review queue with a structured UI that shows the model's last attempt and the validation errors, allowing a human reviewer to correct and submit. Feed those corrections back as few-shot examples.
Multi-model routing: For documents with high extraction complexity (many fields, poor scan quality, non-English), route to a larger/more capable model. For simple extractions, route to a faster, cheaper model. Your validation pipeline already captures the data you need for this routing decision.
Schema-driven UI generation: Your Pydantic models, since they're fully typed with field descriptions, can auto-generate review interfaces and API documentation. Explore libraries like pydantic-to-typescript and FastAPI's automatic OpenAPI docs generation.
Confidence calibration: Extend your extraction models to include a confidence field for uncertain extractions. Train a lightweight classifier on your attempts and error history data to predict which documents are likely to need human review before you even attempt extraction.
The path to reliable AI agents isn't about finding a model that never makes mistakes. It's about building systems that catch mistakes, correct them automatically when possible, and escalate gracefully when not. The contract you've built here is that infrastructure.
Learning Path: RAG & AI Agents