My first LangGraph production agent silently discarded every job for weeks. The agent was supposed to process user requests from Telegram, break them down into sub-tasks, and execute them across multiple steps, storing intermediate results. Instead, it would process the first step, then restart from scratch on the next invocation, losing all prior context. The problem wasn't in the agent logic itself, but in how I was attempting to persist its state.
I burned through three distinct LangGraph checkpointing strategies before I found one that reliably worked for stateful agents in production. Each failure taught me a critical lesson about the assumptions LangGraph makes and the realities of deploying multi-step AI agents.
The Silent State Schema Mismatch
My initial approach was simple: use LangGraph's built-in SqliteSaver. It seemed robust enough for a single-instance deployment on an Oracle Cloud VM. The agent's graph defined a state, let's call it AgentState, with fields like user_id: str, request_id: str, task_list: list[str], and current_step: int.
class AgentState(TypedDict):
user_id: str
request_id: str
task_list: list[str]
current_step: int
# ... more fields added later
The agent would receive a message, initialize AgentState, and kick off the graph. Subsequent messages from the same user_id were supposed to resume the existing state.
The problem started when I needed to add a new field to AgentState, say tool_output: dict. I updated the TypedDict, redeployed the agent, and expected it to pick up where it left off. It didn't. Existing conversations would restart. New conversations worked fine.
I debugged for days, tracing SqliteSaver calls, checking the database directly. The checkpoint table in SQLite stores a JSON blob of the state. What I found was insidious: LangGraph's SqliteSaver does not perform schema migration or even warn about schema mismatches. When an older checkpoint (without tool_output) was loaded into an agent expecting the new AgentState schema, the TypedDict instantiation would silently drop any fields not present in the loaded JSON, and also fail to initialize new fields with default values if they weren't explicitly handled.
The agent would load a partial state, proceed as if it were complete, and then fail downstream because tool_output was missing. Or, worse, it would just restart the entire process because a critical flag like current_step was reset due to the partial load. This wasn't an error; it was a silent data loss. My solution was a manual, painful process: dump the SQLite checkpoints, manually migrate the JSON, and re-insert. This was not scalable.
The Checkpoint Corruption Lottery
After the SqliteSaver debacle, I moved to a custom OracleCloudObjectStorageSaver. My agents run on Oracle Cloud Infrastructure (OCI), and object storage is cheap and highly available. I implemented a BaseCheckpointSaver subclass that would serialize the state to JSON and upload it to an OCI object storage bucket, using the thread_id as the object name.
class OracleCloudObjectStorageSaver(BaseCheckpointSaver):
def __init__(self, bucket_name: str, namespace: str, object_storage_client):
self.bucket_name = bucket_name
self.namespace = namespace
self.client = object_storage_client
def get(self, thread_id: str) -> Optional[Checkpoint]:
try:
response = self.client.get_object(self.namespace, self.bucket_name, thread_id)
data = json.loads(response.data.content.decode('utf-8'))
return Checkpoint(**data)
except Exception as e:
# Log and return None if object not found or corrupted
print(f"Error loading checkpoint {thread_id}: {e}")
return None
def put(self, thread_id: str, checkpoint: Checkpoint) -> None:
data = json.dumps(checkpoint, default=str) # default=str handles datetime objects
self.client.put_object(
self.namespace, self.bucket_name, thread_id, data.encode('utf-8')
)
This seemed more robust. I had full control over serialization and deserialization. I could add versioning to my state objects and handle migrations explicitly.
Then came the checkpoint corruption. Occasionally, an agent would fail to load its state, reporting a JSON decoding error. The get_object call would return valid data, but json.loads would throw. Upon inspection, the JSON files in OCI object storage were truncated or malformed.
The root cause was concurrency. My agents are stateless in themselves, running as serverless functions or on VMs that can scale. Multiple invocations for the same thread_id could happen almost simultaneously, especially if a user sent rapid-fire messages. If two put operations happened concurrently, one might overwrite the other partially, or a read might occur while a write was in progress, leading to corrupted JSON. OCI Object Storage provides eventual consistency, but not strong consistency for overwrites.
My initial fix was to add a retry mechanism with exponential backoff for put operations. This reduced the frequency of corruption but didn't eliminate it. The problem was fundamental: a simple overwrite model for state in a highly concurrent environment is a race condition waiting to happen.
The Atomic Update Pattern: Versioning and Conditional Writes
The solution that finally stabilized LangGraph stateful agents in production involved two key components: explicit state versioning and conditional writes.
Instead of just storing the Checkpoint object, I wrapped it in a custom envelope that included a version number.
class VersionedCheckpoint(TypedDict):
version: int
checkpoint: Checkpoint
When loading a checkpoint, I would read the version field. When writing, I would increment it. The crucial part was the conditional write. OCI Object Storage, like S3, supports conditional requests using If-Match or If-None-Match headers, based on the ETag of the object. This allows for optimistic locking.
My put operation was modified to:
1. Read the current object (if it exists) to get its ETag and current version.
2. Increment the version number.
3. Attempt to write the new object, including the new version, conditionally. If the ETag of the object on the server doesn't match the ETag I read in step 1, it means another process modified the object. The write fails.
4. If the write fails due to an ETag mismatch, retry the entire process (read, increment, write) a few times.
class OracleCloudObjectStorageAtomicSaver(BaseCheckpointSaver):
def __init__(self, bucket_name: str, namespace: str, object_storage_client):
self.bucket_name = bucket_name
self.namespace = namespace
self.client = object_storage_client
self.max_retries = 5
def get(self, thread_id: str) -> Optional[Checkpoint]:
try:
response = self.client.get_object(self.namespace, self.bucket_name, thread_id)
data = json.loads(response.data.content.decode('utf-8'))
# Expecting VersionedCheckpoint structure
versioned_data = VersionedCheckpoint(**data)
return Checkpoint(**versioned_data['checkpoint'])
except Exception as e:
print(f"Error loading checkpoint {thread_id}: {e}")
return None
def put(self, thread_id: str, checkpoint: Checkpoint) -> None:
for attempt in range(self.max_retries):
current_etag = None
current_version = 0
# Try to get current object and ETag for conditional write
try:
response = self.client.get_object(self.namespace, self.bucket_name, thread_id)
current_etag = response.headers.get('etag')
existing_data = json.loads(response.data.content.decode('utf-8'))
current_version = existing_data.get('version', 0)
except Exception as e:
# Object might not exist, or other transient error. Proceed with no ETag.
print(f"No existing object or error getting ETag for {thread_id}: {e}")
new_version = current_version + 1
versioned_checkpoint = VersionedCheckpoint(version=new_version, checkpoint=checkpoint)
data_to_write = json.dumps(versioned_checkpoint, default=str)
try:
headers = {'If-Match': current_etag} if current_etag else {}
self.client.put_object(
self.namespace, self.bucket_name, thread_id, data_to_write.encode('utf-8'),
opc_meta={'version': str(new_version)}, # Store version in metadata too
**headers
)
return # Success
except Exception as e:
if "412 Precondition Failed" in str(e) and attempt < self.max_retries - 1:
print(f"Precondition failed for {thread_id}, retrying (attempt {attempt+1})...")
time.sleep(2 ** attempt) # Exponential backoff
else:
raise # Re-raise if max retries reached or other error
raise Exception(f"Failed to save checkpoint for {thread_id} after {self.max_retries} attempts.")
This pattern ensures that only one write operation succeeds at a time for a given thread_id. If multiple agents try to update the same state concurrently, only the one whose If-Match header correctly identifies the current state's ETag will succeed. Others will fail and retry, eventually picking up the newly written state and applying their updates on top of it. This effectively serializes concurrent updates to the same checkpoint.
This atomic update pattern, combined with explicit state versioning and schema management, finally provided the stability needed for production LangGraph agents. My agents now reliably maintain state across multiple steps, even under concurrent load, whether they're routing requests to Groq for fast initial responses or to Claude for complex reasoning. The cost of OCI Object Storage for this is negligible, typically under $5/month for hundreds of thousands of checkpoints.
Lessons Learned for Stateful AI Agents
1. LangGraph's SqliteSaver is for single-process, non-evolving state. It's fine for demos, but not for production where state schemas change or concurrency is a factor.
2. Explicitly manage your state schema. TypedDict is a compile-time hint, not a runtime validator or migrator. Implement your own versioning and migration logic for your state objects.
3. Concurrency is a state killer. Any shared state in a distributed or concurrent system needs an atomic update mechanism. Simple overwrites lead to silent data corruption.
4. Leverage cloud primitives. Object storage with conditional writes (If-Match / ETag) is a powerful, cost-effective primitive for optimistic locking. Don't reinvent the wheel with complex distributed locks unless absolutely necessary.
5. Monitor and log everything. The silent failures were the hardest to debug. Extensive logging of checkpoint loads, saves, versions, and retries is crucial.
My current agents, handling everything from customer support routing to internal data analysis, now run with this pattern. The initial pain of three rewrites was worth the stability and confidence it brought.
Frequently Asked Questions
Q: Why not use a proper database like PostgreSQL with row-level locking for checkpointing?
A: A full relational database adds operational overhead (management, backups, scaling) and cost. For simple key-value state, object storage with conditional writes provides sufficient atomicity and is orders of magnitude cheaper and simpler to operate at scale for this specific use case. My current setup costs under $5/month.
Q: How do you handle schema migrations for AgentState with this approach?
A: When loading a VersionedCheckpoint, I check the version field. If the loaded version is older than the agent's current expected schema version, I apply explicit migration functions (e.g., adding default values for new fields, transforming old field names) before instantiating the AgentState TypedDict.
**Q: What if an agent crashes during a put operation, leaving a corrupted checkpoint?**
A: The put operation is designed to be idempotent and resilient. If a crash occurs mid-write, the next get operation will either retrieve the last successfully written checkpoint (if the partial write didn't overwrite the ETag) or fail to parse the JSON, triggering a retry or a fresh start. The conditional write helps prevent partial writes from corrupting a valid previous state.
Q: Does this approach add significant latency to each agent step?
A: Each get and put operation involves network calls to OCI Object Storage. For typical agent steps, this adds 50-200ms of latency per state access, which is acceptable for most conversational AI applications where LLM calls dominate latency (e.g., Groq 100ms, Claude 1-5s). For extremely high-throughput, low-latency scenarios, an in-memory cache with eventual consistency might be layered on top.