Introduction to Structured Output Validation with LangGraph
When building applications with Large Language Models (LLMs), one of the most persistent challenges developers face is getting the model to return data in a consistent, predictable format. LLMs are inherently probabilistic—they generate text based on likelihoods, not guarantees. This is where structured output validation becomes critical, and LangGraph provides powerful primitives to handle this elegantly.
LangGraph, built on top of LangChain, is a framework for building stateful, multi-actor applications with LLMs. It treats your application as a graph of nodes and edges, where each node performs a computation and edges determine the flow. Within this graph-based architecture, structured output validation ensures that data passing between nodes conforms to expected schemas, enabling reliable, production-grade AI workflows.
What Is Structured Output Validation?
Structured output validation is the process of constraining an LLM's response to match a predefined schema—typically defined using Pydantic models, JSON Schema, or TypedDicts—and then validating that the output actually conforms to that schema. If validation fails, the system can retry, self-correct, or route to an error-handling path.
In the context of LangGraph, this means:
- Defining a schema that represents the expected shape of the LLM's output.
- Instructing the LLM to produce output matching that schema (via function calling, JSON mode, or structured output methods).
- Validating the output against the schema at runtime.
- Handling validation failures through graph edges that route to retry or correction nodes.
Why Structured Output Validation Matters
Without structured output validation, your LLM application is fragile. Consider a customer support agent that needs to extract a user's issue, priority level, and category from a conversation. If the LLM returns free-form text, downstream code that tries to parse priority as an enum or category as a specific string will break unpredictably.
Here are the key reasons structured output validation matters:
- Reliability: Downstream systems can trust the data shape they receive.
- Composability: In a LangGraph workflow, each node can assume the input it receives matches a schema, making the graph robust.
- Error recovery: When validation fails, LangGraph's conditional edges let you route to a correction node rather than crashing.
- Type safety: IDE autocompletion, static analysis, and runtime checks all benefit from well-defined schemas.
- Testing: You can mock nodes with schema-valid data, making your graph testable.
Setting Up Your Environment
Before diving into code, let's set up the environment. You'll need Python 3.10+, LangGraph, LangChain, and Pydantic installed.
pip install langgraph langchain-openai pydantic
Set your OpenAI API key as an environment variable:
export OPENAI_API_KEY="your-api-key-here"
Defining Schemas with Pydantic
Pydantic is the most common way to define schemas in the LangChain/LangGraph ecosystem. It provides declarative validation, custom validators, and automatic JSON Schema generation. Let's define a schema for a customer support ticket extraction task.
from pydantic import BaseModel, Field, field_validator
from typing import Literal, Optional
from datetime import datetime
class SupportTicket(BaseModel):
"""A customer support ticket extracted from a conversation."""
customer_name: str = Field(
description="The full name of the customer reporting the issue."
)
issue_summary: str = Field(
description="A concise one-sentence summary of the customer's issue."
)
priority: Literal["low", "medium", "high", "critical"] = Field(
description="The urgency level of the ticket."
)
category: Literal["billing", "technical", "account", "general"] = Field(
description="The department that should handle this ticket."
)
affected_product: Optional[str] = Field(
default=None,
description="The specific product or feature affected, if mentioned."
)
extracted_at: datetime = Field(
default_factory=datetime.now,
description="Timestamp when the ticket was extracted."
)
@field_validator("issue_summary")
@classmethod
def summary_must_be_concise(cls, v: str) -> str:
if len(v.split()) > 30:
raise ValueError(
"issue_summary must be 30 words or fewer. "
f"Got {len(v.split())} words."
)
return v
Notice the field_validator decorator. This lets you enforce custom business rules beyond simple type checking—in this case, ensuring the summary stays concise. When the LLM produces output that violates this rule, Pydantic raises a ValidationError, which LangGraph can catch and handle.
Basic Structured Output with LangChain LLMs
LangChain's LLM wrappers provide a with_structured_output method that instructs the model to produce output matching a schema. Under the hood, this uses the model's native function calling or JSON mode capabilities.
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4o", temperature=0)
# Bind the schema to the LLM
structured_llm = llm.with_structured_output(SupportTicket)
conversation = """
Customer: Hi, my name is Sarah Johnson. I've been trying to log into
my account for the past hour but I keep getting a 500 error.
This is urgent because I need to submit a report by end of day!
Support Agent: I understand your frustration, Sarah. Let me look
into this right away.
"""
ticket = structured_llm.invoke(
f"Extract a support ticket from this conversation:\n\n{conversation}"
)
print(ticket)
# Output: SupportTicket(
# customer_name='Sarah Johnson',
# issue_summary='Customer cannot log into account due to recurring 500 error.',
# priority='high',
# category='technical',
# affected_product=None,
# extracted_at=datetime(...)
# )
This works well for simple cases, but in a real application, you need error handling, retries, and the ability to route based on validation outcomes. That's where LangGraph comes in.
Building a LangGraph Workflow with Validation
Let's build a LangGraph workflow that extracts a support ticket, validates it, and includes a self-correction loop for when validation fails. This is where structured output validation truly shines.
Defining the Graph State
First, define the state that flows through the graph. This state holds the conversation, the extracted ticket, validation errors, and a retry counter.
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, END
import operator
class GraphState(TypedDict):
conversation: str
ticket: Optional[SupportTicket]
validation_errors: Annotated[list[str], operator.add]
retry_count: int
max_retries: int
The Annotated[list[str], operator.add] annotation tells LangGraph to concatenate lists when merging state updates from multiple nodes, rather than overwriting them. This is useful for accumulating error messages across retries.
Creating the Extraction Node
The extraction node calls the LLM with the structured output method. If the LLM itself raises an error (e.g., it returns malformed JSON), we catch it and record the error.
from pydantic import ValidationError
def extract_ticket_node(state: GraphState) -> dict:
"""Attempt to extract a structured support ticket from the conversation."""
errors = state.get("validation_errors", [])
retry_count = state.get("retry_count", 0)
# Build a prompt that includes previous errors for self-correction
base_prompt = (
"You are a support ticket extraction assistant. "
"Extract a structured support ticket from the following conversation. "
"Be precise and follow the schema exactly.\n\n"
f"Conversation:\n{state['conversation']}"
)
if errors:
error_context = "\n\n".join(f"- {e}" for e in errors[-3:])
base_prompt += (
f"\n\nPrevious attempts failed with these errors:\n{error_context}\n"
"Please correct these issues and try again."
)
try:
ticket = structured_llm.invoke(base_prompt)
return {"ticket": ticket, "retry_count": retry_count + 1}
except Exception as e:
return {
"validation_errors": [f"LLM extraction error: {str(e)}"],
"retry_count": retry_count + 1,
"ticket": None,
}
Creating the Validation Node
The validation node explicitly re-validates the extracted ticket. While with_structured_output already performs validation, having a separate validation node gives you a clear point in the graph to apply additional business logic checks.
def validate_ticket_node(state: GraphState) -> dict:
"""Validate the extracted ticket against the schema and business rules."""
ticket = state.get("ticket")
if ticket is None:
return {"validation_errors": ["No ticket was extracted."]}
# Re-validate using Pydantic (catches any edge cases)
try:
# Convert to dict and back to trigger full validation
validated = SupportTicket(**ticket.model_dump())
# Additional business rule: critical tickets must have a product
if validated.priority == "critical" and not validated.affected_product:
return {
"validation_errors": [
"Critical priority tickets must specify an affected_product."
]
}
return {"ticket": validated, "validation_errors": []}
except ValidationError as e:
error_messages = [f"{err['loc']}: {err['msg']}" for err in e.errors()]
return {"validation_errors": error_messages}
Defining Conditional Routing
Now we need conditional edges that route based on whether validation passed or failed. If validation fails and we haven't exceeded the retry limit, we loop back to extraction. Otherwise, we end.
def should_retry_or_finish(state: GraphState) -> str:
"""Determine the next node based on validation results."""
errors = state.get("validation_errors", [])
retry_count = state.get("retry_count", 0)
max_retries = state.get("max_retries", 3)
if not errors:
return "end"
if retry_count >= max_retries:
return "end"
return "retry"
Assembling the Graph
Now let's wire everything together into a complete LangGraph workflow.
from langgraph.graph import START
# Build the graph
workflow = StateGraph(GraphState)
# Add nodes
workflow.add_node("extract", extract_ticket_node)
workflow.add_node("validate", validate_ticket_node)
# Add edges
workflow.add_edge(START, "extract")
workflow.add_edge("extract", "validate")
# Conditional edge after validation
workflow.add_conditional_edges(
"validate",
should_retry_or_finish,
{
"retry": "extract",
"end": END,
},
)
# Compile
app = workflow.compile()
Running the Workflow
Let's run the workflow with a sample conversation and inspect the results.
conversation = """
Customer: Hello, I'm Michael Chen. I was charged twice for my
Pro subscription this month and I need a refund immediately.
This is affecting my business operations.
Support Agent: I apologize for the inconvenience, Michael.
Let me check your billing right away.
"""
initial_state: GraphState = {
"conversation": conversation,
"ticket": None,
"validation_errors": [],
"retry_count": 0,
"max_retries": 3,
}
final_state = app.invoke(initial_state)
if final_state.get("validation_errors"):
print("Validation failed after retries:")
for err in final_state["validation_errors"]:
print(f" - {err}")
else:
ticket = final_state["ticket"]
print("Successfully extracted ticket:")
print(f" Customer: {ticket.customer_name}")
print(f" Summary: {ticket.issue_summary}")
print(f" Priority: {ticket.priority}")
print(f" Category: {ticket.category}")
print(f" Product: {ticket.affected_product}")
Using TypedDict and JSON Schema Alternatives
While Pydantic is the most popular choice, LangGraph also supports TypedDict and JSON Schema for defining structured outputs. TypedDict is lighter weight but lacks runtime validation. JSON Schema works well when you're integrating with external systems that already define schemas.
from typing import TypedDict
class SimpleTicket(TypedDict):
"""A lightweight ticket schema without runtime validation."""
customer_name: str
issue_summary: str
priority: str
category: str
# Use with structured output
simple_llm = llm.with_structured_output(SimpleTicket)
For JSON Schema, you can pass a dictionary directly:
json_schema = {
"title": "SupportTicket",
"description": "A customer support ticket.",
"type": "object",
"properties": {
"customer_name": {"type": "string"},
"issue_summary": {"type": "string"},
"priority": {
"type": "string",
"enum": ["low", "medium", "high", "critical"],
},
},
"required": ["customer_name", "issue_summary", "priority"],
}
schema_llm = llm.with_structured_output(json_schema)
Advanced Pattern: Parallel Validation with Multiple Schemas
In more complex workflows, you might want to extract multiple structured outputs in parallel—for example, extracting both a ticket and a sentiment analysis from the same conversation. LangGraph supports this through fan-out patterns.
class SentimentAnalysis(BaseModel):
"""Sentiment analysis result for a conversation."""
sentiment: Literal["positive", "neutral", "negative", "frustrated"]
confidence: float = Field(ge=0.0, le=1.0, description="Confidence score 0-1")
key_phrases: list[str] = Field(description="Notable phrases indicating sentiment")
def extract_sentiment_node(state: GraphState) -> dict:
"""Extract sentiment analysis from the conversation."""
sentiment_llm = llm.with_structured_output(SentimentAnalysis)
prompt = (
"Analyze the sentiment of this customer support conversation:\n\n"
f"{state['conversation']}"
)
try:
result = sentiment_llm.invoke(prompt)
return {"sentiment": result}
except Exception as e:
return {"sentiment": None, "validation_errors": [str(e)]}
# Extended state to include sentiment
class ExtendedGraphState(TypedDict):
conversation: str
ticket: Optional[SupportTicket]
sentiment: Optional[SentimentAnalysis]
validation_errors: Annotated[list[str], operator.add]
retry_count: int
max_retries: int
You can then add both extraction nodes to the graph and use a fan-out from the START node to run them in parallel, followed by a fan-in to merge results.
Best Practices for Structured Output Validation
1. Write Descriptive Field Descriptions
The LLM uses field descriptions to understand what to put in each field. Vague descriptions lead to poor extractions. Be explicit about format, constraints, and examples.
# Bad
class Bad(BaseModel):
name: str
# Good
class Good(BaseModel):
name: str = Field(
description="The customer's full legal name, including first and last name. "
"Example: 'Jane Marie Smith'"
)
2. Use Enums and Literals for Constrained Values
Whenever a field should only accept specific values, use Literal or Enum. This dramatically reduces invalid outputs.
from enum import Enum
class Priority(str, Enum):
LOW = "low"
MEDIUM = "medium"
HIGH = "high"
CRITICAL = "critical"
3. Always Include a Retry Loop
Never assume the LLM will produce valid output on the first try. A retry loop with error feedback is essential for production reliability. Feed the validation errors back into the prompt so the model can self-correct.
4. Set Temperature to 0 for Extraction Tasks
For structured extraction, you want deterministic outputs. Set temperature=0 to reduce randomness.
llm = ChatOpenAI(model="gpt-4o", temperature=0)
5. Keep Schemas Focused
Avoid creating massive schemas with dozens of fields. If you need to extract a lot of information, break it into multiple smaller extraction steps within your graph. Smaller schemas are easier for the LLM to fill correctly and easier to validate.
6. Use Custom Validators for Business Logic
Pydantic validators let you encode domain-specific rules. Use them to catch issues that type checking alone can't detect.
@field_validator("priority")
@classmethod
def critical_requires_context(cls, v, info):
values = info.data
if v == "critical" and not values.get("affected_product"):
raise ValueError(
"Critical priority requires an affected_product to be specified."
)
return v
7. Log Validation Failures for Monitoring
In production, track how often validation fails and which fields are most problematic. This data helps you improve your prompts and schemas over time.
import logging
logger = logging.getLogger(__name__)
def validate_ticket_node(state: GraphState) -> dict:
ticket = state.get("ticket")
if ticket is None:
logger.warning("No ticket extracted", extra={"retry": state.get("retry_count")})
return {"validation_errors": ["No ticket was extracted."]}
# ... rest of validation
8. Consider Using Structured Output Methods Over JSON Mode
LangChain's with_structured_output with Pydantic uses function calling by default, which is more reliable than raw JSON mode. JSON mode ensures valid JSON but doesn't guarantee the JSON matches your schema. Function calling enforces the schema at the API level.
Handling Edge Cases
Empty or Ambiguous Conversations
Not every conversation will contain enough information to fill all required fields. Handle this by making non-essential fields optional and adding a confidence or completeness indicator.
class SupportTicket(BaseModel):
customer_name: Optional[str] = Field(
default=None,
description="Customer's full name. Leave null if not mentioned."
)
issue_summary: str = Field(description="Summary of the issue.")
completeness: Literal["partial", "complete"] = Field(
description="Whether all required information was present in the conversation."
)
Graceful Degradation
When retries are exhausted, don't just fail silently. Return a partial result or escalate to a human reviewer.
def should_retry_or_finish(state: GraphState) -> str:
errors = state.get("validation_errors", [])
retry_count = state.get("retry_count", 0)
max_retries = state.get("max_retries", 3)
if not errors:
return "end"
if retry_count >= max_retries:
# Escalate to human review
return "escalate"
return "retry"
def escalate_node(state: GraphState) -> dict:
"""Escalate failed extraction to human review."""
logger.error(
"Ticket extraction failed after max retries",
extra={
"conversation": state["conversation"],
"errors": state.get("validation_errors", []),
}
)
return {"ticket": None}
Testing Your Validation Logic
Test your schemas and validation nodes independently of the LLM. This ensures your validation logic is correct before you account for LLM variability.
import pytest
from pydantic import ValidationError
def test_valid_ticket():
ticket = SupportTicket(
customer_name="Test User",
issue_summary="Login button not working.",
priority="high",
category="technical",
)
assert ticket.priority == "high"
def test_invalid_priority():
with pytest.raises(ValidationError):
SupportTicket(
customer_name="Test User",
issue_summary="Login button not working.",
priority="urgent", # Not a valid priority
category="technical",
)
def test_summary_too_long():
long_summary = " ".join(["word"] * 50)
with pytest.raises(ValidationError):
SupportTicket(
customer_name="Test User",
issue_summary=long_summary,
priority="low",
category="general",
)
Conclusion
Structured output validation is a cornerstone of building reliable LLM applications, and LangGraph provides the perfect architecture for implementing it with retry loops, conditional routing, and stateful error handling. By defining clear Pydantic schemas, writing descriptive field annotations, implementing self-correcting retry loops, and following best practices like using enums for constrained values and logging validation failures, you can build AI workflows that produce consistent, trustworthy data. The combination of LangGraph's graph-based execution model and Pydantic's robust validation creates a powerful foundation for production-grade applications where data integrity matters as much as the intelligence of the model itself. Start simple with a single extraction node and validation edge, then gradually add complexity like parallel extraction, custom business validators, and human escalation paths as your application's needs grow.