State Typing and Validation
Type-safe state management is crucial for building reliable LangGraph applications. This guide covers TypedDict, Pydantic models, validation strategies, and type safety best practices.
Note:
StateGraphsupportsTypedDict, dataclass, and Pydantic state schemas. In contrast, LangChaincreate_agentcustom state schemas must beTypedDict.
TypedDict Basics
TypedDict provides type hints for dictionary-based state without runtime validation.
Python TypedDict
from typing import TypedDict, Annotated, Literal
from langgraph.graph.message import add_messages
from langchain_core.messages import BaseMessage
class BasicState(TypedDict):
"""Basic state schema with type hints."""
query: str
results: list[dict]
status: Literal["pending", "processing", "complete"]
count: int
Key features:
- Provides IDE autocomplete and type checking
- No runtime validation by default
- Lightweight (no extra dependencies)
- Works with mypy and other type checkers
Required vs Optional Fields
from typing import TypedDict, NotRequired, Required
class StateWithOptional(TypedDict, total=False):
"""All fields optional by default."""
query: str
results: list[dict]
metadata: dict
class MixedState(TypedDict):
"""Mix required and optional fields."""
query: Required[str] # Required
results: list[dict] # Required (default)
metadata: NotRequired[dict] # Optional
Nested TypedDict
class UserInfo(TypedDict):
"""Nested user information."""
user_id: str
username: str
email: str
class RequestMetadata(TypedDict):
"""Request metadata."""
timestamp: float
ip_address: str
class ComplexState(TypedDict):
"""State with nested structures."""
user: UserInfo
metadata: RequestMetadata
messages: list[str]
Pydantic Models
Pydantic provides runtime validation, serialization, and advanced features.
Basic Pydantic State
from pydantic import BaseModel, Field
from typing import Literal
class PydanticState(BaseModel):
"""Pydantic state with validation."""
query: str = Field(..., min_length=1, max_length=500)
results: list[dict] = Field(default_factory=list)
status: Literal["pending", "processing", "complete"] = "pending"
count: int = Field(default=0, ge=0)
class Config:
"""Pydantic config."""
validate_assignment = True # Validate on attribute assignment
Key features:
- Runtime validation
- Default values with
Field - Validators and constraints
- JSON serialization
- Data parsing and coercion
Field Validation
from pydantic import BaseModel, Field, validator, field_validator
class ValidatedState(BaseModel):
"""State with field validation."""
email: str = Field(..., pattern=r'^[\w\.-]+@[\w\.-]+\.\w+$')
age: int = Field(..., ge=0, le=150)
score: float = Field(..., ge=0.0, le=1.0)
tags: list[str] = Field(..., min_items=1, max_items=10)
@field_validator('email')
@classmethod
def validate_email(cls, v: str) -> str:
"""Custom email validation."""
if not v or '@' not in v:
raise ValueError('Invalid email address')
return v.lower()
@field_validator('tags')
@classmethod
def validate_tags(cls, v: list[str]) -> list[str]:
"""Validate and normalize tags."""
return [tag.strip().lower() for tag in v if tag.strip()]
Model Validation
from pydantic import BaseModel, model_validator
class CrossFieldState(BaseModel):
"""State with cross-field validation."""
start_date: str
end_date: str
duration_days: int
@model_validator(mode='after')
def check_dates_consistent(self) -> 'CrossFieldState':
"""Validate dates are consistent."""
from datetime import datetime
start = datetime.fromisoformat(self.start_date)
end = datetime.fromisoformat(self.end_date)
actual_duration = (end - start).days
if actual_duration != self.duration_days:
raise ValueError(
f'Duration mismatch: {actual_duration} vs {self.duration_days}'
)
return self
Pydantic with LangGraph
from typing import TypedDict
from pydantic import BaseModel, Field
from langgraph.graph import MessagesState
# Option 1: Pydantic model converted to dict
class PydanticModel(BaseModel):
"""Pydantic model for validation."""
query: str
threshold: float = Field(ge=0.0, le=1.0)
class GraphState(MessagesState):
"""MessagesState for LangGraph."""
config: dict # Store Pydantic model as dict
def node_with_pydantic(state: GraphState) -> dict:
"""Node using Pydantic for validation."""
# Parse and validate
config = PydanticModel(**state["config"])
# Use validated config
print(f"Query: {config.query}")
print(f"Threshold: {config.threshold}")
return {}
# Option 2: Hybrid approach
class ValidatedConfig(BaseModel):
"""Validated configuration."""
max_iterations: int = Field(ge=1, le=100)
temperature: float = Field(ge=0.0, le=2.0)
class HybridState(MessagesState):
"""MessagesState with Pydantic field."""
config: ValidatedConfig # Use Pydantic directly
# Usage
state = HybridState(
messages=[],
config=ValidatedConfig(max_iterations=10, temperature=0.7)
)
TypeScript Zod Validation
Zod provides runtime validation in TypeScript.
Basic Zod Schema with StateSchema
import { StateSchema } from "@langchain/langgraph";
import { z } from "zod/v4";
// Define state using StateSchema with Zod types directly
const AppState = new StateSchema({
query: z.string().min(1).max(500),
results: z.array(z.record(z.string(), z.any())).default([]),
status: z.enum(["pending", "processing", "complete"]).default("pending"),
count: z.number().int().nonnegative().default(0),
});
// Extract TypeScript types
type AppStateType = typeof AppState.State;
type AppStateUpdate = typeof AppState.Update;
Validation in Nodes
import { z } from "zod/v4";
const EmailSchema = z.string().email();
const AgeSchema = z.number().int().min(0).max(150);
function validateNode(state: any): Partial<any> {
// Validate input
const email = EmailSchema.parse(state.email);
const age = AgeSchema.parse(state.age);
// Process with validated data
console.log(`Email: ${email}, Age: ${age}`);
return {};
}
Custom Validation
import { z } from "zod/v4";
const DateValidationSchema = z.object({
startDate: z.string(),
endDate: z.string(),
durationDays: z.number(),
}).refine(
(data) => {
const start = new Date(data.startDate);
const end = new Date(data.endDate);
const actualDuration = Math.floor(
(end.getTime() - start.getTime()) / (1000 * 60 * 60 * 24)
);
return actualDuration === data.durationDays;
},
{
message: "Duration must match the difference between dates",
}
);
Advanced Type Patterns
Generic State
from typing import Generic, TypeVar
from langgraph.graph import MessagesState
T = TypeVar('T')
class GenericState(MessagesState, Generic[T]):
"""Generic state for reusable patterns."""
data: T
status: str
# Use with specific types
StringState = GenericState[str]
DictState = GenericState[dict]
Union Types
from typing import TypedDict, Union, Literal
class TextResult(TypedDict):
"""Text result type."""
type: Literal["text"]
content: str
class ImageResult(TypedDict):
"""Image result type."""
type: Literal["image"]
url: str
alt_text: str
class MultiModalState(TypedDict):
"""State with union types."""
results: list[Union[TextResult, ImageResult]]
# Type-safe access
def process_results(state: MultiModalState):
for result in state["results"]:
if result["type"] == "text":
# Type checker knows this is TextResult
print(result["content"])
elif result["type"] == "image":
# Type checker knows this is ImageResult
print(result["url"], result["alt_text"])
Tagged Unions
from typing import TypedDict, Literal, Union
class SuccessResult(TypedDict):
"""Success result."""
status: Literal["success"]
data: dict
class ErrorResult(TypedDict):
"""Error result."""
status: Literal["error"]
error_message: str
error_code: int
Result = Union[SuccessResult, ErrorResult]
class ResultState(TypedDict):
"""State with tagged union."""
result: Result
def handle_result(state: ResultState):
"""Type-safe result handling."""
result = state["result"]
if result["status"] == "success":
# TypedDict narrows to SuccessResult
print(result["data"])
else:
# TypedDict narrows to ErrorResult
print(f"Error {result['error_code']}: {result['error_message']}")
State Validation Strategies
Input Validation
from typing import TypedDict
from pydantic import BaseModel, Field, ValidationError
class InputModel(BaseModel):
"""Validate input."""
query: str = Field(..., min_length=1)
max_results: int = Field(default=10, ge=1, le=100)
class GraphState(TypedDict):
"""Graph state."""
query: str
max_results: int
results: list[dict]
def validate_input(raw_input: dict) -> GraphState:
"""Validate and convert input."""
try:
validated = InputModel(**raw_input)
return {
"query": validated.query,
"max_results": validated.max_results,
"results": []
}
except ValidationError as e:
raise ValueError(f"Invalid input: {e}")
# Usage
try:
state = validate_input({"query": "test", "max_results": 5})
except ValueError as e:
print(f"Validation failed: {e}")
Runtime State Validation
from langgraph.graph import MessagesState
class ValidatedState(MessagesState):
"""State with runtime validation."""
iteration: int
max_iterations: int
def validate_state(state: ValidatedState) -> None:
"""Runtime state validation."""
if state["iteration"] > state["max_iterations"]:
raise ValueError(
f"Iteration {state['iteration']} exceeds max {state['max_iterations']}"
)
if not state["messages"]:
raise ValueError("Messages cannot be empty")
def safe_node(state: ValidatedState) -> dict:
"""Node with state validation."""
validate_state(state)
# Process with validated state
return {"iteration": state["iteration"] + 1}
Conditional Validation
from pydantic import BaseModel, field_validator
class ConditionalState(BaseModel):
"""State with conditional validation."""
mode: Literal["development", "production"]
api_key: str | None = None
@field_validator('api_key')
@classmethod
def validate_api_key(cls, v, info):
"""Validate API key in production."""
if info.data.get('mode') == 'production' and not v:
raise ValueError('API key required in production mode')
return v
Type Safety Best Practices
1. Use Literal for Enums
# ✅ Good: Type-safe with Literal
from typing import Literal
class GoodState(TypedDict):
status: Literal["pending", "processing", "complete"]
# ❌ Bad: Any string accepted
class BadState(TypedDict):
status: str
2. Annotate Lists and Dicts
# ✅ Good: Specific types
class GoodState(TypedDict):
results: list[dict[str, Any]]
metadata: dict[str, str]
# ❌ Bad: Untyped
class BadState(TypedDict):
results: list
metadata: dict
3. Use NotRequired for Optional Fields
from typing import TypedDict, NotRequired
# ✅ Good: Clear optional fields
class GoodState(TypedDict):
query: str # Required
results: NotRequired[list[dict]] # Optional
# ❌ Bad: Ambiguous
class BadState(TypedDict):
query: str
results: list[dict] | None # Nullable but still required
4. Document State Fields
class DocumentedState(TypedDict):
"""Well-documented state.
Attributes:
query: User's search query
results: Search results from API
status: Current processing status
timestamp: When the query was received
"""
query: str
results: list[dict]
status: Literal["pending", "processing", "complete"]
timestamp: float
5. Validate Early
def node_with_validation(state: MyState) -> dict:
"""Validate at node entry."""
# Validate immediately
if not state["query"]:
raise ValueError("Query cannot be empty")
if state["iteration"] > 100:
raise ValueError("Too many iterations")
# Continue with valid state
...
Testing Type Safety
MyPy Configuration
# mypy.ini
[mypy]
python_version = 3.11
warn_return_any = True
warn_unused_configs = True
disallow_untyped_defs = True
disallow_any_generics = True
check_untyped_defs = True
Type Checking Tests
from typing import TypedDict
class TestState(TypedDict):
value: str
count: int
def test_type_safety():
"""Test type safety with mypy."""
# This should pass mypy
state: TestState = {"value": "test", "count": 1}
# This should fail mypy
# state: TestState = {"value": 123, "count": "wrong"} # Type error
def test_required_fields():
"""Test required fields."""
# Missing required field - mypy error
# state: TestState = {"value": "test"} # Error: missing 'count'
# All required fields present - OK
state: TestState = {"value": "test", "count": 1}
assert state["count"] == 1
Runtime Validation Tests
from pydantic import BaseModel, ValidationError
import pytest
class ValidatedModel(BaseModel):
value: str
count: int
def test_pydantic_validation():
"""Test Pydantic runtime validation."""
# Valid input
model = ValidatedModel(value="test", count=1)
assert model.count == 1
# Invalid input
with pytest.raises(ValidationError):
ValidatedModel(value=123, count="wrong")
# Missing field
with pytest.raises(ValidationError):
ValidatedModel(value="test")
Common Type Issues
Issue: Any Type
from typing import Any
# ❌ Bad: Loses type safety
class BadState(TypedDict):
data: Any # Could be anything
# ✅ Good: Specific type
class GoodState(TypedDict):
data: dict[str, str] | list[int] # Clear alternatives
Issue: Mutable Defaults
# ❌ Bad: Mutable default
class BadState(TypedDict):
items: list[str] = [] # Shared between instances!
# ✅ Good: Use default_factory with Pydantic
from pydantic import BaseModel, Field
class GoodState(BaseModel):
items: list[str] = Field(default_factory=list)
Issue: Type Coercion
from pydantic import BaseModel
class State(BaseModel):
count: int
# Pydantic coerces types
state = State(count="123") # Converts to int(123)
# To prevent coercion
class StrictState(BaseModel):
count: int
class Config:
strict = True # No coercion
# Now this raises ValidationError
# state = StrictState(count="123")
Performance Considerations
TypedDict vs Pydantic
import timeit
# TypedDict (no validation)
from typing import TypedDict
class TypedDictState(TypedDict):
value: str
count: int
# Pydantic (with validation)
from pydantic import BaseModel
class PydanticState(BaseModel):
value: str
count: int
# Benchmark
typeddict_time = timeit.timeit(
lambda: {"value": "test", "count": 1},
number=100000
)
pydantic_time = timeit.timeit(
lambda: PydanticState(value="test", count=1),
number=100000
)
# TypedDict is ~10x faster (no validation overhead)
# Use TypedDict for performance-critical paths
# Use Pydantic for validation-critical paths
Validation Caching
from functools import lru_cache
from pydantic import BaseModel
class Config(BaseModel):
setting1: str
setting2: int
@lru_cache(maxsize=100)
def get_validated_config(setting1: str, setting2: int) -> Config:
"""Cache validated configs."""
return Config(setting1=setting1, setting2=setting2)
# Repeated calls use cache
config1 = get_validated_config("test", 1)
config2 = get_validated_config("test", 1) # From cache
assert config1 is config2
Migration Path
From Untyped to TypedDict
# Before: Untyped dict
def old_node(state: dict) -> dict:
query = state["query"] # No type checking
return {"results": []}
# After: TypedDict
class State(TypedDict):
query: str
results: list[dict]
def new_node(state: State) -> dict:
query = state["query"] # Type checked!
return {"results": []}
From TypedDict to Pydantic
# Before: TypedDict
class OldState(TypedDict):
email: str
age: int
# After: Pydantic with validation
from pydantic import BaseModel, Field
class NewState(BaseModel):
email: str = Field(..., pattern=r'^[\w\.-]+@[\w\.-]+\.\w+$')
age: int = Field(..., ge=0, le=150)
Additional Resources
- Pydantic Documentation: https://docs.pydantic.dev/
- Zod Documentation: https://zod.dev/
- MyPy Documentation: https://mypy.readthedocs.io/
- LangGraph State Documentation: https://docs.langchain.com/oss/python/langgraph/graph-api