tools-only State Schema Patterns
Common state schema patterns for LangGraph applications with examples for different use cases.
Chat Application State
Basic Chat State
from langgraph.graph import MessagesState
class ChatState(MessagesState):
"""State for basic chat application."""
TypeScript:
import { StateSchema, MessagesValue } from "@langchain/langgraph";
const ChatState = new StateSchema({
messages: MessagesValue,
});
Chat with Session Management
class SessionChatState(MessagesState):
"""Chat state with session tracking."""
session_id: str
user_id: str
context: dict
last_activity: float
Multi-Turn Conversation State
class ConversationState(MessagesState):
"""State for multi-turn conversations with context."""
conversation_id: str
turn_count: int
context_summary: str
topics: list[str]
Research Agent State
Basic Research State
class ResearchState(TypedDict):
"""State for research agent."""
query: str
search_results: list[dict]
processed_sources: list[str]
summary: str
citations: list[str]
Multi-Source Research State
def extend_list(left: list, right: list) -> list:
"""Extend list reducer."""
return left + right
class MultiSourceResearchState(TypedDict):
"""Research from multiple sources."""
query: str
# Sources
web_results: Annotated[list[dict], extend_list]
academic_results: Annotated[list[dict], extend_list]
news_results: Annotated[list[dict], extend_list]
# Processing
processed_count: int
total_sources: int
# Output
synthesis: str
confidence: float
Research with Validation State
class ValidatedResearchState(TypedDict):
"""Research state with fact-checking."""
query: str
# Research phase
raw_results: list[dict]
facts: list[dict]
# Validation phase
validated_facts: list[dict]
contradictions: list[dict]
confidence_scores: dict[str, float]
# Output
final_report: str
Workflow Orchestration State
Task Workflow State
from typing import Literal
class TaskWorkflowState(TypedDict):
"""State for task-based workflow."""
task: str
status: Literal["pending", "in_progress", "completed", "failed"]
current_step: str
steps_completed: list[str]
results: dict[str, Any]
errors: list[str]
Approval Workflow State
class ApprovalWorkflowState(TypedDict):
"""Workflow with human approval steps."""
request: dict
# Processing
draft: str
status: Literal["draft", "pending_approval", "approved", "rejected"]
# Approval tracking
approver: str
approval_timestamp: float
rejection_reason: str | None
# Retry
revision_count: int
max_revisions: int
Pipeline State
class PipelineState(TypedDict):
"""Multi-stage pipeline state."""
input_data: dict
# Stage outputs
stage1_output: dict
stage2_output: dict
stage3_output: dict
# Tracking
current_stage: int
stages_completed: list[str]
stage_errors: dict[str, str]
# Final
final_output: dict
pipeline_success: bool
Tool-Calling Agent State
Basic Tool-Calling State
from langgraph.graph import MessagesState
class ToolCallingState(MessagesState):
"""State for agent with tools."""
# Tool tracking
tool_calls_made: list[str]
tool_results: dict[str, Any]
# Control
should_continue: bool
max_iterations: int
current_iteration: int
Advanced Tool State
class AdvancedToolState(MessagesState):
"""Tool-calling with error handling and retries."""
# Tool execution
pending_tools: list[dict]
completed_tools: list[dict]
failed_tools: list[dict]
# Retry logic
retry_count: dict[str, int]
max_retries: int
# Results
tool_outputs: dict[str, Any]
final_answer: str
RAG (Retrieval-Augmented Generation) State
Basic RAG State
class RAGState(TypedDict):
"""State for RAG application."""
query: str
# Retrieval
retrieved_docs: list[dict]
relevance_scores: list[float]
# Generation
context: str
response: str
sources: list[str]
Multi-Index RAG State
class MultiIndexRAGState(TypedDict):
"""RAG with multiple vector stores."""
query: str
# Retrieval from multiple indexes
docs_index1: list[dict]
docs_index2: list[dict]
docs_index3: list[dict]
# Reranking
reranked_docs: list[dict]
# Generation
context: str
response: str
confidence: float
Conversational RAG State
class ConversationalRAGState(MessagesState):
"""RAG with conversation history."""
# Context
conversation_context: str
# Retrieval
query_rephrased: str
retrieved_docs: list[dict]
# Generation
response: str
sources: list[str]
Data Processing State
Batch Processing State
class BatchProcessingState(TypedDict):
"""State for batch data processing."""
items: list[dict]
# Processing
processed_items: Annotated[list[dict], extend_list]
failed_items: Annotated[list[dict], extend_list]
# Progress
total_count: int
processed_count: int
failed_count: int
# Results
results: dict
processing_complete: bool
ETL Pipeline State
class ETLState(TypedDict):
"""Extract-Transform-Load pipeline."""
# Extract
source_data: list[dict]
extraction_timestamp: float
# Transform
transformed_data: list[dict]
transformation_rules: dict
validation_errors: list[str]
# Load
load_status: Literal["pending", "loading", "completed", "failed"]
records_loaded: int
load_errors: list[str]
Code Generation State
Code Generation State
class CodeGenState(TypedDict):
"""State for code generation agent."""
task_description: str
language: str
# Generation
generated_code: str
test_cases: list[str]
# Validation
syntax_valid: bool
tests_passed: bool
# Iteration
revision_count: int
feedback: str
Code Review State
class CodeReviewState(TypedDict):
"""State for code review agent."""
code: str
language: str
# Analysis
issues: list[dict]
suggestions: list[dict]
complexity_score: float
# Review
review_comments: str
approval_status: Literal["approved", "changes_requested", "rejected"]
Error Recovery State
Retry State
class RetryState(TypedDict):
"""State with retry logic."""
task: str
# Execution
result: Any | None
error: str | None
# Retry tracking
attempt_count: int
max_attempts: int
last_error: str
# Status
status: Literal["pending", "running", "success", "failed"]
Fallback State
class FallbackState(TypedDict):
"""State with fallback strategies."""
task: str
# Primary approach
primary_result: Any | None
primary_error: str | None
# Fallback approaches
fallback_attempted: list[str]
fallback_result: Any | None
# Final
final_result: Any
strategy_used: str
Complex Multi-Agent State
Supervisor State
class SupervisorState(MessagesState):
"""Supervisor-subagent pattern state."""
# Routing
next: Literal["agent1", "agent2", "agent3", "FINISH"]
current_agent: str
# Tracking
agent_history: list[str]
iteration_count: int
# Shared context
context: dict
Router State
class RouterState(MessagesState):
"""Router pattern state."""
# Routing decision
route: Literal["category1", "category2", "category3"]
confidence: float
# Execution
result: Any
Orchestrator-Worker State
class OrchestratorState(TypedDict):
"""Orchestrator-worker pattern state."""
task: str
# Decomposition
subtasks: list[dict]
# Execution
results: Annotated[list[str], extend_list]
# Progress
completed_count: int
total_count: int
# Aggregation
final_result: str
Custom Reducer Patterns
List Extension
def extend_list(left: list, right: list) -> list:
"""Extend list without duplicates."""
return left + right
class ListState(TypedDict):
"""State with list extension."""
items: Annotated[list[str], extend_list]
Dictionary Merging
def merge_dicts(left: dict, right: dict) -> dict:
"""Merge dictionaries with right overriding left."""
return {**left, **right}
class DictState(TypedDict):
"""State with dictionary merging."""
context: Annotated[dict, merge_dicts]
Set Union
def union_sets(left: set, right: set) -> set:
"""Union of sets."""
return left | right
class SetState(TypedDict):
"""State with set union."""
tags: Annotated[set[str], union_sets]
Conditional Update
def conditional_update(left: Any, right: Any | None) -> Any:
"""Update only if right is not None."""
return right if right is not None else left
class ConditionalState(TypedDict):
"""State with conditional updates."""
status: Annotated[str, conditional_update]
State Schema Best Practices
1. Use TypedDict for Type Safety
# ✅ Good: TypedDict provides type safety
class MyState(TypedDict):
field: str
# ❌ Bad: Plain dict loses type information
my_state = {"field": "value"}
2. Document State Fields
class DocumentedState(TypedDict):
"""Well-documented state schema."""
# Input
query: str # User's search query
# Processing
results: list[dict] # Search results from API
# Output
response: str # Generated response
3. Use Literal for Routing Fields
# ✅ Good: Literal provides type safety for routing
class GoodState(TypedDict):
next: Literal["agent1", "agent2", "FINISH"]
# ❌ Bad: String without type constraint
class BadState(TypedDict):
next: str # Could be anything
4. Choose Appropriate Reducers
# ✅ Good: Use MessagesState for chat history
class ChatState(MessagesState):
pass
# ❌ Bad: No reducer, loses previous messages
messages: list
5. Include Metadata for Debugging
class DebugState(TypedDict):
"""State with debugging metadata."""
# Core fields
task: str
result: str
# Debugging metadata
start_time: float
end_time: float
iteration_count: int
errors: list[str]
6. Keep State Flat When Possible
# ✅ Good: Flat structure
class FlatState(TypedDict):
user_id: str
user_name: str
user_email: str
# ⚠️ Consider: Nested structure (use sparingly)
class NestedState(TypedDict):
user: dict # Less type-safe
7. Use Optional for Nullable Fields
from typing import Optional
class OptionalFieldState(TypedDict):
required_field: str
optional_field: Optional[str] # Can be None
Additional Resources
- LangGraph State Documentation: https://docs.langchain.com/oss/python/langgraph/graph-api
- Message Reducers: https://docs.langchain.com/oss/python/langgraph/use-graph-api