Type-safe LLM responses in LangChain with Pydantic

If you’ve ever parsed JSON from an LLM with regex and a prayer, you know why this langchain type safe llm responses pydantic tutorial exists. Structured output turns probabilistic text into data your code can trust — no post-hoc cleaning, no silent schema drift. LangChain’s with_structured_output and Pydantic v2 make this straightforward, but the defaults hide sharp edges. Here’s the path from “it works on my prompt” to “it works in production.”

Why structured output changes the contract

LLMs are stochastic. Without constraints, a model asked for “a JSON object with name and email” returns markdown fences, extra commentary, missing fields, or hallucinated keys. Pydantic models move the contract from documentation to enforcement: validation fails fast, types propagate through your editor, and refactoring catches breaking changes before deploy.

LangChain wraps this in with_structured_output, which injects a schema into the system prompt and parses the response. The abstraction is thin — you still control the model, the prompt, and the error handling. That’s the right level: opinionated defaults, escape hatches everywhere.

Install the right pieces

pip install langchain-core langchain-openai pydantic

Use Pydantic v2 (the default since 2.0). V1 is end-of-life and lacks model_dump_json, model_validate_json, and the performance gains that matter at scale. If you’re pinned to v1 for legacy code, migrate first — the API surface is similar but the internals differ enough to cause subtle bugs.

Define schemas that survive contact with reality

Start with a model that mirrors your domain, not the LLM’s whims.

from pydantic import BaseModel, Field, EmailStr, field_validator
from typing import Optional, Literal
from datetime import datetime


class SupportTicket(BaseModel):
    ticket_id: str = Field(pattern=r"^TKT-\d{6}$")
    category: Literal["billing", "technical", "account", "other"]
    priority: Literal["low", "medium", "high", "critical"]
    summary: str = Field(min_length=10, max_length=200)
    customer_email: EmailStr
    created_at: datetime = Field(default_factory=datetime.utcnow)
    tags: list[str] = Field(default_factory=list, max_length=10)

    @field_validator("summary")
    @classmethod
    def no_markdown(cls, v: str) -> str:
        if any(c in v for c in "`*#[]"):
            raise ValueError("summary must be plain text")
        return v.strip()

Notes on the choices above:

• pattern on ticket_id catches format drift early.
• Literal restricts enums to known values — the model can’t invent “urgent” when you only handle “high” and “critical.”
• EmailStr validates format without regex in your code.
• The summary validator strips markdown the model loves to emit.
• default_factory for created_at ensures server-side authority; never trust the model for timestamps.

Avoid Optional for required fields. If the business logic requires a value, make it required and handle the validation error. Optional with a default of None pushes the problem downstream.

Wire it to the model

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
structured_llm = llm.with_structured_output(SupportTicket, method="function_calling")

method="function_calling" uses OpenAI’s function calling API — the most reliable path for OpenAI models. For other providers, method="json_mode" requests JSON output and parses it, but you lose schema enforcement in the prompt. Some providers (Anthropic, local models via Ollama) support tool calling; check the provider’s LangChain integration for the correct method name.

The temperature=0 matters. Structured output is a classification task disguised as generation. Non-zero temperature introduces variance that breaks schemas.

Invoke and handle the happy path

prompt = """Extract a support ticket from the user's message.
Return only the structured object. No markdown, no explanation."""

messages = [
    SystemMessage(content=prompt),
    HumanMessage(content="I'm jane@acme.com and my invoice INV-4421 was charged twice. Ticket TKT-123456."),
]

ticket: SupportTicket = structured_llm.invoke(messages)
print(ticket.model_dump_json(indent=2))

Output:

{
  "ticket_id": "TKT-123456",
  "category": "billing",
  "priority": "high",
  "summary": "Invoice INV-4421 charged twice",
  "customer_email": "jane@acme.com",
  "created_at": "2024-01-15T14:32:11.123456",
  "tags": ["billing", "duplicate-charge"]
}

The return type is SupportTicket, not dict or Any. Your editor knows the fields. Mypy passes. Refactoring category to issue_type breaks at compile time, not in logs at 2 AM.

Validation errors are control flow, not exceptions

with_structured_output raises OutputParserException on parse failure. Catch it, log the raw response, and decide: retry, escalate, or return a fallback.

from langchain_core.exceptions import OutputParserException
from tenacity import retry, stop_after_attempt, wait_exponential

@retry(
    wait=wait_exponential(multiplier=1, min=2, max=10),
    stop=stop_after_attempt(3),
    reraise=True,
)
def extract_ticket(messages: list) -> SupportTicket:
    try:
        return structured_llm.invoke(messages)
    except OutputParserException as e:
        # Log raw response for debugging — it's in e.llm_output
        logger.warning("Structured output parse failed", raw=e.llm_output)
        raise

# Usage
try:
    ticket = extract_ticket(messages)
except OutputParserException:
    # Fallback: queue for human review
    queue_for_review(messages[-1].content)

Don’t swallow the exception. The raw LLM output (e.llm_output) is gold for prompt iteration. Store it.

Streaming structured output — partial validation

Streaming token-by-token into a Pydantic model is a trap. You get partial JSON that fails validation until complete. LangChain’s astream with with_structured_output yields the final parsed object only after the full response arrives — no partials.

If you need progressive UI updates, stream raw tokens and parse incrementally yourself, or use a two-stage approach: stream a summary field first, then fetch the full object.

# This yields the complete SupportTicket only at the end
async for chunk in structured_llm.astream(messages):
    # chunk is SupportTicket (final) — no intermediate states
    pass

For true incremental parsing, consider pydantic-partial or a custom parser that accumulates JSON and validates on } boundaries. But weigh the complexity: most UX needs are served by streaming a “thinking” indicator, then showing the complete validated object.

Nested models and recursive schemas

Real domains nest. Pydantic handles it; the LLM needs explicit guidance.

class LineItem(BaseModel):
    sku: str
    qty: int = Field(gt=0)
    unit_price_cents: int = Field(ge=0)


class Order(BaseModel):
    order_id: str
    items: list[LineItem] = Field(min_length=1)
    total_cents: int = Field(ge=0)

    @model_validator(mode="after")
    def totals_match(self) -> "Order":
        calculated = sum(i.qty * i.unit_price_cents for i in self.items)
        if calculated != self.total_cents:
            raise ValueError(f"total_cents {self.total_cents} != calculated {calculated}")
        return self

The model_validator (v2’s replacement for root_validator) runs after all fields parse. It catches the model hallucinating a total that doesn’t match line items — a common failure mode when the LLM computes rather than extracts.

Prompt for nested structures explicitly:

prompt = """Extract an order with line items.
Each item needs sku, qty, unit_price_cents.
The total_cents must equal sum(qty * unit_price_cents).
Return only the JSON object."""

Enum drift and versioning

Enums in Literal are a contract. When the business adds “refund” as a category, you have two choices:

1. Add to the Literal — redeploy the parser. The model may still output old values until you update the prompt examples.
2. Use a string with a validator — accept anything, normalize in code.

class TicketCategory(str):
    BILLING = "billing"
    TECHNICAL = "technical"
    ACCOUNT = "account"
    OTHER = "other"

class SupportTicketV2(BaseModel):
    category: str
    
    @field_validator("category")
    @classmethod
    def normalize_category(cls, v: str) -> str:
        mapping = {
            "billing": "billing",
            "payments": "billing",
            "refund": "billing",
            "tech": "technical",
            "bug": "technical",
        }
        return mapping.get(v.lower(), "other")

Approach 2 buys flexibility at the cost of static checking. Choose based on how often categories change and whether downstream systems can handle unknown values.

Common pitfalls

The model ignores the schema

Symptom: valid JSON, wrong fields. Cause: the schema wasn’t injected into the prompt. Fix: verify with_structured_output actually sends the schema. For method="function_calling", it does. For method="json_mode", it prepends a schema description — but some models ignore it. Test with a deliberately wrong prompt and confirm validation fails.

DateTime parsing fails silently

Pydantic v2 parses ISO 8601 by default. If the model emits “January 15, 2024”, validation fails. Either constrain the prompt (“ISO 8601 format only”) or add a BeforeValidator to parse common formats.

from pydantic import BeforeValidator
from typing import Annotated
from dateutil import parser as dateparser

FlexibleDatetime = Annotated[datetime, BeforeValidator(lambda v: dateparser.parse(v) if isinstance(v, str) else v)]

class Event(BaseModel):
    occurred_at: FlexibleDatetime

Large schemas exceed context

A 50-field model with descriptions burns 2k+ tokens. If you hit context limits, split: extract a minimal schema first, then enrich with a second call. Or use model.model_json_schema() to generate a compact schema description manually.

The model wraps output in markdown

Even with function calling, some models emit ```json fences. The parser handles this for OpenAI function calling, but json_mode can choke. Strip fences in a pre-parser if needed:

def strip_fences(text: str) -> str:
    if text.startswith("```"):
        text = text.split("\n", 1)[1]
        text = text.rsplit("```", 1)[0]
    return text

Testing structured output

Unit test the schema, not the LLM.

import pytest
from pydantic import ValidationError


def test_ticket_validation():
    # Valid
    ticket = SupportTicket(
        ticket_id="TKT-123456",
        category="billing",
        priority="high",
        summary="Invoice charged twice",
        customer_email="jane@acme.com",
    )
    assert ticket.category == "billing"

    # Invalid category
    with pytest.raises(ValidationError):
        SupportTicket(
            ticket_id="TKT-123456",
            category="refund",  # not in Literal
            priority="high",
            summary="Invoice charged twice",
            customer_email="jane@acme.com",
        )

    # Invalid email
    with pytest.raises(ValidationError):
        SupportTicket(
            ticket_id="TKT-123456",
            category="billing",
            priority="high",
            summary="Invoice charged twice",
            customer_email="not-an-email",
        )

Integration test the prompt + model combo with a few golden examples. Store the expected SupportTicket objects as JSON fixtures. Run nightly — model behavior drifts.

When to skip structured output

Not every call needs it. Use raw generation for:

• Creative writing, summarization, chat
• Exploratory prompts where the schema isn’t stable
• Calls where latency budget is tight and the schema adds 500ms+

Structured output adds a parsing round-trip (or function calling overhead). Profile your pipeline. If the parser is 5% of latency, it’s free. If it’s 40%, reconsider.

Putting it in production

The pattern that scales:

1. Schema as code — Pydantic models in a shared package, versioned with your API.
2. Prompt as code — Templates in version control, not string literals.
3. Observability — Log every validation failure with the raw response. Alert on failure rate spikes.
4. Fallback chain — Structured output → JSON mode → raw text → human queue.
5. Routing — If you hit rate limits or provider degradation on one model, fail over to another that supports the same schema. An inference gateway like n4n.ai handles this routing transparently while preserving your structured output contract across 240+ models.

The last point matters: your schema is a contract. The model fulfilling it is an implementation detail. Build the pipeline so swapping gpt-4o-mini for claude-3-haiku or a fine-tuned Llama 3 changes one config line, not your parsing logic.

Start small. Pick one extraction task that currently uses regex. Replace it with a Pydantic model and with_structured_output. Ship it. Watch the validation errors — they’re your test suite. Iterate the prompt until failures drop near zero. Then move to the next task. Type safety compounds.