My first LangGraph agent silently dropped 80% of its jobs for three weeks. The logs showed success, the agent responded, but the core task—generating a specific output—never materialized. The problem wasn't the LLM, nor the tools. It was a state schema mismatch, a silent killer in a system designed for stateful persistence. This wasn't a "hello world" problem; this was a production agent, processing real user requests for AI-generated content, running on Oracle Cloud Infrastructure (OCI) with Groq and Claude models.
The Silent State Schema Mismatch
The initial LangGraph setup used a TypedDict for its state. Simple, clear, and Pythonic.
class AgentState(TypedDict):
input: str
intermediate_results: List[str]
final_output: Optional[str]
error: Optional[str]
The agent's nodes would update intermediate_results and eventually final_output. The problem appeared when I introduced a new node that needed to track a specific counter, say, retry_count: int. I updated the TypedDict:
class AgentState(TypedDict):
input: str
intermediate_results: List[str]
final_output: Optional[str]
error: Optional[str]
retry_count: int # New field
I deployed the new version. Existing checkpoints, created with the old schema, were loaded. New jobs started. Everything seemed fine. The agent ran, the retry_count was incremented within the node, but when the state was checkpointed and reloaded for the next step, retry_count was gone. It was silently discarded.
LangGraph's default MemorySaver (and by extension, the SQLSaver I was using) serializes the state. When deserializing, if the schema changes, fields present in the serialized data but not in the current TypedDict are often ignored or dropped, depending on the exact deserialization mechanism. More critically, new fields in the TypedDict that aren't in the old serialized state are simply initialized to their default (or None if Optional). My retry_count was always 0 on reload, effectively creating an infinite loop for certain error conditions.
The fix was to explicitly manage schema evolution. I switched to a Pydantic model for state, which offers better validation and error handling on deserialization.
from pydantic import BaseModel, Field
from typing import List, Optional
class AgentState(BaseModel):
input: str
intermediate_results: List[str] = Field(default_factory=list)
final_output: Optional[str] = None
error: Optional[str] = None
retry_count: int = 0 # Default value for new fields
version: int = 1 # Schema versioning
@classmethod
def from_old_state(cls, old_state: dict):
# Migration logic here if needed
return cls(**old_state)
When loading a checkpoint, I'd check state.version. If it was older, I'd run a migration function. This added overhead but prevented silent data loss. The SQLSaver now stored a JSON blob, and Pydantic handled the validation on load. This immediately surfaced errors instead of silently dropping data.
Checkpoint Corruption and Race Conditions
The second rewrite came from checkpoint corruption. My agents run on OCI Container Instances, processing requests from Telegram and WhatsApp via a custom API gateway. Each user interaction could trigger a new graph run or resume an existing one. The SQLSaver was backed by an Oracle Autonomous Database.
The issue: multiple concurrent updates to the same checkpoint. Imagine a user sends a message. Agent starts. User sends another message before the first one completes. The second message triggers a new graph run, but LangGraph, seeing an existing thread ID, tries to load the same checkpoint.
If SQLSaver attempts to write a checkpoint while another process is still reading or writing it, you get database-level race conditions. Sometimes, a UNIQUE constraint (USER.LANGGRAPH_CHECKPOINTS_PK) violated error. Other times, a partial write, leading to a json.decoder.JSONDecodeError: Expecting value: line 1 column 1 (char 0) on the next load. The metadata or state column in the langgraph_checkpoints table would be truncated or malformed.
The SQLSaver uses a simple INSERT OR REPLACE for updates. This is not atomic for complex state objects. A proper solution requires pessimistic locking or a versioning system at the database level.
My workaround involved two steps:
1. Application-level locking: Before loading or updating a checkpoint, acquire a distributed lock using Redis (OCI Cache with Redis). This ensures only one agent instance can touch a specific thread ID's checkpoint at a time. This added ~50ms latency per checkpoint operation but eliminated corruption.
import redis
import os
REDIS_HOST = os.environ.get("REDIS_HOST")
REDIS_PORT = int(os.environ.get("REDIS_PORT", 6379))
REDIS_DB = int(os.environ.get("REDIS_DB", 0))
redis_client = redis.Redis(host=REDIS_HOST, port=REDIS_PORT, db=REDIS_DB)
def acquire_lock(thread_id: str, timeout: int = 60):
lock_name = f"langgraph_lock:{thread_id}"
lock = redis_client.lock(lock_name, timeout=timeout)
if not lock.acquire(blocking=True, timeout=10): # Try to acquire for 10s
raise TimeoutError(f"Could not acquire lock for thread {thread_id}")
return lock
def release_lock(lock):
lock.release()
# Usage:
# with acquire_lock(thread_id) as lock:
# # Load/save checkpoint
2. Idempotent updates: The agent's nodes were refactored to be more idempotent. Instead of just appending to intermediate_results, each update would check if the new data was already present or if the state was already in a desired terminal condition. This reduced the impact of retries or duplicate executions.
This significantly improved stability. The TimeoutError from the lock acquisition became a clear signal of contention, which I could handle gracefully (e.g., tell the user "Please wait, your previous request is still processing").
The Stable Pattern: Explicit State Transitions and Sub-Graphs
The final rewrite, and the one that truly made LangGraph stateful agents production-ready for me, involved a fundamental shift in how I thought about state and transitions. Instead of a monolithic graph, I started using sub-graphs and explicit state transition nodes.
My agents often involve a sequence like:
1. Parse user input.
2. Route to an appropriate tool/model (Groq for fast, Claude for complex).
3. Execute tool.
4. Generate response.
5. Optionally, ask for clarification or re-route.
Initially, this was one large graph with many conditional edges. The state object grew unwieldy, and debugging became a nightmare. The "current step" was implicit in the graph's execution path.
The stable pattern:
1. Define clear AgentState enums for major phases.
from enum import Enum
class AgentPhase(str, Enum):
INPUT_PARSING = "input_parsing"
TOOL_ROUTING = "tool_routing"
TOOL_EXECUTION = "tool_execution"
RESPONSE_GENERATION = "response_generation"
CLARIFICATION = "clarification"
FINISHED = "finished"
ERROR = "error"
class AgentState(BaseModel):
# ... existing fields ...
current_phase: AgentPhase = AgentPhase.INPUT_PARSING
last_tool_output: Optional[str] = None
# ...
2. Each major phase is a sub-graph or a single node.
Instead of a single app.compile(), I'd have input_parser_graph, tool_router_node, tool_executor_graph, etc.
The main graph then orchestrates these.
3. A dedicated "transition" node.
After each major step, a node explicitly updates current_phase in the state. This node also contains the logic for conditional transitions.
def transition_node(state: AgentState) -> AgentState:
if state.error:
state.current_phase = AgentPhase.ERROR
elif state.final_output:
state.current_phase = AgentPhase.FINISHED
elif state.last_tool_output and not state.final_output:
state.current_phase = AgentPhase.RESPONSE_GENERATION
# ... more complex logic ...
return state
# In the main graph:
# graph.add_node("transition", transition_node)
# graph.add_edge("tool_executor_graph_end", "transition")
# graph.add_conditional_edges(
# "transition",
# lambda state: state.current_phase,
# {
# AgentPhase.RESPONSE_GENERATION: "response_generator_node",
# AgentPhase.FINISHED: END,
# AgentPhase.ERROR: END,
# # ...
# }
# )
This pattern made the graph's flow explicit and debuggable. If an agent got stuck, I could inspect the current_phase in the checkpoint and immediately know where it failed. It also allowed for easier checkpointing between major phases, rather than relying on LangGraph's internal step-by-step saves. I could even implement custom retry logic for specific phases. For instance, if TOOL_EXECUTION failed, I could increment retry_count and transition back to TOOL_ROUTING with a different model.
This modularity, combined with the Pydantic state schema and distributed locking, transformed my LangGraph agents from fragile prototypes into robust production systems capable of handling thousands of concurrent user interactions across multiple messaging platforms. The cost of these three rewrites was significant in engineering hours, but the stability and reliability gained were essential for shipping production AI agents with zero VC funding. Every dropped job was a direct hit to user trust and my ability to generate revenue.
Frequently Asked Questions
Q: Why not use LangGraph's built-in checkpoint_saver with a custom serde?
A: While possible, the SQLSaver's default json.dumps and json.loads don't handle schema evolution gracefully. A custom serde would still need to implement migration logic, and Pydantic provides a more robust, battle-tested framework for data validation and schema management than rolling your own.
Q: How much overhead did the Redis lock add to each checkpoint operation?
A: On OCI Cache with Redis, a SETNX (acquire) and DEL (release) operation typically adds 5-15ms. For a full checkpoint load-and-save cycle, including network latency to the database, the total overhead was around 50ms per checkpoint operation. This was acceptable for my use case where agent steps are often LLM calls taking hundreds of milliseconds.
Q: What if the Redis lock itself fails or the agent crashes while holding a lock?
A: Redis locks should always have an expiry (timeout parameter in redis_client.lock). If an agent crashes, the lock will automatically expire after the set duration (e.g., 60 seconds), preventing permanent deadlocks. This means a job might be delayed, but not permanently stuck.
Q: Why not use a more sophisticated workflow engine like Apache Airflow or Temporal for state?
A: LangGraph provides a Python-native, LLM-centric way to define agentic workflows, tightly integrated with LangChain's ecosystem. For workflows heavily reliant on LLM reasoning and tool use, LangGraph's graph-based approach is more intuitive than traditional DAG orchestrators. The goal was to make LangGraph itself robust, not replace it.