Module 3.4: Structured Outputs & Schema Adherence

Curriculum Alignment: docs/plan/03_phase3_modular_chaining_lcel.md
Topic Scope: Structured Outputs, Pydantic v2 Schemas, .with_structured_output() Method, Computer-Readable Contracts
Level: Advanced AI Engineering / Architecture

---

1. The Fragility of Unstructured Natural Language

In enterprise backends, software systems cannot reliably consume raw markdown, conversational chatter, or unvalidated JSON strings. A model outputting "Here is your JSON: ```json { 'status': 'ok' } ```" requires fragile regex extraction and manual string trimming that frequently breaks in production.

To bridge non-deterministic natural language generation with deterministic software systems, we require Structured Outputs:

• The LLM is constrained to emit strictly valid JSON conforming to an exact JSON Schema.
• The runtime deserializes the output directly into strongly typed Pydantic v2 domain models.
• Any deviation from the schema (missing keys, invalid enum values, extra unexpected properties) immediately triggers validation exceptions before corrupted data enters your database or message queue.

---

2. The .with_structured_output() Pattern

Rather than writing custom prompt instructions pleading with the model to "output only valid JSON", modern frontier models natively support Structured Outputs via Function Calling or JSON Schema mode.

LangChain wraps this capability into the universal method:

$$\text{structured\_model}=\text{model.with\_structured\_output}(\text{PydanticSchema})$$

---

3. Production Python Implementation: The TaskPlan Extractor

Here is a complete production example demonstrating the extraction of unstructured project notes into an immutable TaskPlan domain object:

"""
Schema-Enforced Extraction via with_structured_output()
Demonstrates strict Pydantic v2 typing and validation constraints.
"""
from enum import Enum
from typing import List
from pydantic import BaseModel, ConfigDict, Field
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

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

class ActionableStep(BaseModel):
    """Atomic step within a task plan."""
    model_config = ConfigDict(frozen=True, extra="forbid")
    
    step_number: int = Field(ge=1, description="Sequential index of the step.")
    description: str = Field(description="Clear, actionable command of what to execute.")
    estimated_hours: float = Field(gt=0.0, le=40.0, description="Estimated effort in hours.")
    requires_approval: bool = Field(description="True if step modifies production state.")

class TaskPlan(BaseModel):
    """Root computer-readable task plan contract."""
    model_config = ConfigDict(frozen=True, extra="forbid")
    
    title: str = Field(min_length=5, max_length=120, description="Concise plan title.")
    priority: TaskPriority = Field(description="Assessed priority level.")
    architecture_domain: str = Field(description="Domain impacted (e.g. Auth, Caching, DB).")
    steps: list[ActionableStep] = Field(min_length=1, description="Ordered list of execution steps.")
    risk_summary: str = Field(description="Summary of primary technical risks.")

def create_task_extraction_chain():
    """Builds an LCEL chain enforcing TaskPlan structured output."""
    model = ChatOpenAI(model="gpt-4o-mini", temperature=0.0)
    
    # Bind structured output to model
    structured_model = model.with_structured_output(TaskPlan)
    
    prompt = ChatPromptTemplate.from_messages([
        ("system", """You are an automated Technical Project Manager.
Extract unstructured architectural meeting notes into an immutable, structured TaskPlan.
Ensure all action items are assigned realistic hour estimates and priority levels."""),
        ("human", "Unstructured Meeting Notes:\n<notes>\n{meeting_notes}\n</notes>")
    ])
    
    # Composed LCEL chain directly emits a validated TaskPlan object
    return prompt | structured_model

Usage and Execution:

chain = create_task_extraction_chain()
result = chain.invoke({
    "meeting_notes": """
    Yesterday the team agreed to migrate Redis caching from cluster A to cluster B.
    We need to first provision the new Redis instance in VPC-2 (takes ~4 hours).
    Then configure cross-region replication and verify ping latency (2 hours).
    Finally, perform DNS switchover during the maintenance window (1 hour, needs prod approval).
    """
})

# Result is a true Pydantic instance, fully typed with autocomplete
print(type(result))  # <class 'TaskPlan'>
print(result.title)  # "Redis Cache Cluster Migration"
print(result.priority)  # TaskPriority.HIGH
print(len(result.steps))  # 3

Conceptual Mindmap: Structured Outputs

---

4. Curated Reading & Canonical References

Resource	Canonical Reference & Link	Specific Focus Areas
Primary Curriculum Book	AI Agents and Applications (Google Drive)	Part 2, Chapter 4, Section 4.3.4 (p. 103): Converting raw model JSON into typed Python domain objects.
LangChain Core Reference	Structured Outputs Guide	Function calling constraints, strict schema modes, and provider compatibility.
Data Validation Standard	Pydantic v2 Documentation	Type annotations, performance validation, and domain value object immutability.

---

5. Active Recall (Module 3.4 Flashcards)

Structured OutputsClick or press Space to flip ↺

How does .with_structured_output() differ from prompting a model to 'Output JSON'?

Question / Scenario

Structured Outputs • AnswerClick to flip back ↻

Prompting to 'Output JSON' relies purely on semantic compliance; the model often returns markdown codeblocks, explanatory comments, or malformed syntax. .with_structured_output() configures the underlying model API's native function calling or JSON schema mode, mathematically constraining token sampling to valid schema tokens and parsing directly into a Pydantic object.

💡 Architect Takeaway: Never parse raw model JSON manually; use .with_structured_output(Schema).

Technical Explanation

Schema DefenseClick or press Space to flip ↺

Why is ConfigDict(extra='forbid') crucial in production Pydantic extraction schemas?

Question / Scenario

Schema Defense • AnswerClick to flip back ↻

Models frequently hallucinate additional metadata keys, commentary, or debugging flags inside JSON objects. extra='forbid' causes Pydantic to instantly reject unexpected fields, maintaining strict domain contract boundaries and preventing silent schema pollution.

💡 Architect Takeaway: Enforce extra='forbid' on all external LLM ingestion schemas.

Technical Explanation

---

6. Hands-on Engineering Exercises

Exercise 3.4: The Schema-Enforced Extractor Drill

• Goal: Build the foundation for your Schema-Enforced Extractor deliverable.
• Task: Take a raw, messy paragraph describing an infrastructure outage post-mortem.
• Requirements:
  • Define an IncidentPostMortem Pydantic model with root_cause, outage_duration_minutes, services_affected, and action_items.
  • Bind to ChatOpenAI using .with_structured_output(IncidentPostMortem).
  • Prove that executing the chain outputs an immutable, strongly typed instance.