Structured Outputs and JSON Enforcement

In automated data pipelines, an LLM completion is almost never consumed directly by a human. It is consumed by downstream code: Airflow task operators, PySpark DataFrames, or database insertion cursors.

If an LLM returns:

Here is the JSON you requested:
```json
{"status": "SUCCESS"}

I hope this helps!

Your downstream JSON parser (`json.loads()`) will throw a `JSONDecodeError` and crash the pipeline.

**Structured Outputs** is the engineering discipline of forcing LLMs to return 100% syntactically valid JSON conforming strictly to a pre-defined schema.

---

## 1. The 3 Approaches to JSON Generation

```mermaid
flowchart TD
    subgraph Approaches["Approaches to Structured Data"]
        PromptOnly["1. Prompt-Only ('Please output JSON')<br/>❌ 5-15% failure rate (chatty headers, trailing commas)"]
        JSONMode["2. Native JSON Mode (OpenAI / Gemini)<br/>⚠️ Guarantees valid JSON syntax, but may miss specific schema keys"]
        Constrained["3. Strict Schemas / Constrained Logit Masking (Pydantic / Outlines)<br/>✅ 100% Guaranteed JSON matching exact types & required keys"]

        PromptOnly --> JSONMode --> Constrained
    end

    style PromptOnly fill:#991b1b,color:#fff
    style JSONMode fill:#d97706,color:#fff
    style Constrained fill:#047857,color:#fff

2. Constrained Decoding at the Logits Level

How do frontier model APIs guarantee valid JSON without post-processing?

The breakthrough is Logits Masking via Context-Free Grammars (CFG / GBNF):

sequenceDiagram
    participant LLM as Model Logits Core
    participant Masker as Grammar Masker (Outlines/GBNF)
    participant Output as Generated Stream

    LLM->>Masker: Raw logits over 128k vocabulary
    Masker->>Masker: Schema rule: Expected string quote '"'
    Masker->>LLM: Mask all tokens EXCEPT quotes to -infinity (-inf)
    LLM->>Output: Emits '"' with 100% certainty

    Masker->>Masker: Schema rule: Expected key name 'user_id'
    Masker->>LLM: Mask all tokens EXCEPT 'user_id'
    LLM->>Output: Emits 'user_id'

At each token generation step, the grammar engine evaluates the current JSON parse state. Any vocabulary token that would produce invalid JSON is assigned a probability of zero ($-\infty$ logit). It is physically impossible for the model to emit a syntax error!

3. Defining Schemas with Pydantic

In Python data engineering, Pydantic is the gold standard for defining structured models:

from pydantic import BaseModel, Field
from typing import List, Optional
from enum import Enum

class SeverityLevel(str, Enum):
    LOW = "LOW"
    MEDIUM = "MEDIUM"
    HIGH = "HIGH"
    CRITICAL = "CRITICAL"

class DataPipelineError(BaseModel):
    task_id: str = Field(description="Name of the failed Airflow task or Spark job")
    root_cause_summary: str = Field(description="One-sentence technical summary of the failure")
    severity: SeverityLevel = Field(description="Impact level on downstream SLAs")
    affected_tables: List[str] = Field(description="List of target tables that were not updated")
    suggested_fix_sql: Optional[str] = Field(description="Remediation SQL query if applicable", default=None)
    retry_recommended: bool = Field(description="Whether transient failure makes retry viable")

4. Production Robust Extraction Pattern with Self-Healing

Even with structured modes, production ETL pipelines should implement validation and automatic error reflection:

import json
from pydantic import BaseModel, ValidationError

class ExtractedInvoice(BaseModel):
    vendor_name: str
    invoice_number: str
    total_amount: float
    currency: str
    line_items_count: int

def parse_with_reflection(llm_output: str, max_retries: int = 2) -> ExtractedInvoice:
    """
    Attempts to parse LLM output against Pydantic schema.
    If parsing fails, returns error context for self-healing.
    """
    # Strip accidental markdown code fences
    cleaned = llm_output.strip()
    if cleaned.startswith("```json"):
        cleaned = cleaned[7:]
    if cleaned.startswith("```"):
        cleaned = cleaned[3:]
    if cleaned.endswith("```"):
        cleaned = cleaned[:-3]
    cleaned = cleaned.strip()

    try:
        data = json.loads(cleaned)
        return ExtractedInvoice(**data)
    except (json.JSONDecodeError, ValidationError) as err:
        print(f"[Schema Failure]: {err}")
        # In production: pass 'err' back to LLM in turn 2 to self-correct
        raise err

if __name__ == "__main__":
    sample_llm_response = """
    ```json
    {
        "vendor_name": "Databricks Inc.",
        "invoice_number": "INV-2026-9921",
        "total_amount": 14250.75,
        "currency": "USD",
        "line_items_count": 4
    }
    ```
    """

    invoice = parse_with_reflection(sample_llm_response)
    print("Parsed Pydantic Object Successfully:")
    print(f"Vendor : {invoice.vendor_name}")
    print(f"Amount : {invoice.currency} {invoice.total_amount:,.2f}")

Key Takeaways

• Never rely on pure prompt instructions for pipeline-critical JSON.
• Use native Structured Outputs (OpenAI / Gemini response_schema) or constrained logit masking.
• Define schemas with Pydantic: Provides type-checking, serialization, and clean integration into data engineering codebases.