My first LangGraph agent silently discarded every job for weeks. The memory field in my AgentState was defined as a list[str], but my agent was writing a list[dict]. LangGraph's SqliteSaver didn't throw an error; it just truncated the state, leaving an empty list. I only found out when a customer complained their multi-step request never progressed past the first turn. The fix was a one-line schema change, but the cost was hours of debugging and a lost customer. This wasn't the only checkpointing pitfall.
The Silent Schema Mismatch: SqliteSaver's Forgiveness
My initial LangGraph setup used SqliteSaver for checkpointing. It's simple, embedded, and seemed robust enough for early production. The agent's purpose was to process incoming requests from Telegram, break them down, and manage a multi-step conversation. The AgentState looked something like this:
class AgentState(TypedDict):
chat_id: str
thread_id: str
user_input: str
memory: list[str] # The culprit
current_step: int
# ... other fields
The agent's memory field was intended to store a history of conversation snippets. My agent code, however, was designed to store more structured data, like {"role": "user", "content": "..."}. So, instead of list[str], it was pushing list[dict] into memory.
SqliteSaver uses json.dumps to serialize the state into a BLOB. When json.dumps encountered a list[dict] where it expected a list[str] (based on the initial schema it inferred or was given), it didn't fail. Instead, it silently serialized the list[dict] into a string. The deserialization step was the problem. When SqliteSaver loaded the state, it tried to deserialize that string back into a list[str]. Since a string representation of list[dict] is not a valid list[str], it often resulted in an empty list or a malformed object, effectively wiping out the conversation history for that specific field.
The fix was to explicitly define the memory field as list[dict] in AgentState. This highlighted a critical lesson: LangGraph's SqliteSaver is forgiving to a fault. It prioritizes saving something over strict schema validation during serialization, leading to silent data corruption on deserialization. For production, you need explicit validation or a more robust ORM.
Checkpoint Corruption: The Race Condition with SqliteSaver
After fixing the schema, I started seeing sqlite3.OperationalError: database is locked errors. My agents run on Oracle Cloud Infrastructure (OCI) in a serverless function (OCI Functions) environment. Each incoming message triggers a new function invocation. While OCI Functions are stateless, my LangGraph agents needed state. SqliteSaver writes to a file. In a serverless environment, this file needs to be externalized. I used OCI Object Storage to store the SQLite database file, mounting it via FUSE.
The problem: multiple concurrent function invocations could try to write to the same SQLite file simultaneously. Even with FUSE, the underlying sqlite3 library isn't designed for concurrent writes from separate processes without proper locking mechanisms, which FUSE-mounted object storage doesn't natively provide at the database level. This led to checkpoint corruption. A database lock error would leave the SQLite file in an inconsistent state, making it unreadable for subsequent invocations.
The One Pattern That Made Multi-Step Pipelines Stable: Explicit State Transitions
My initial graph design relied heavily on conditional edges that checked the content of the user_input or the presence of certain fields in the state. For example:
def route_next_step(state: AgentState):
if state.get("user_input") == "confirm":
return "confirm_action"
if state.get("task_completed"):
return "notify_user"
return "process_input"
This approach was brittle. If user_input wasn't exactly "confirm", or if task_completed was set but another condition also matched, the agent could loop or jump to the wrong node. The problem was that the state itself wasn't explicitly guiding the transition.
The breakthrough came when I introduced an explicit next_action field in my AgentState and made every node responsible for setting it.
class AgentState(TypedDict):
chat_id: str
thread_id: str
user_input: str
memory: list[dict]
current_step: Literal["start", "process_input", "confirm_action", "notify_user", "end"] # Explicit state
next_action: Literal["process_input", "confirm_action", "notify_user", "end", "wait_for_user"] # Guiding the graph
# ... other fields
Now, my conditional edges became much simpler and more robust:
def route_next_action(state: AgentState):
return state["next_action"]
# In my graph definition:
graph.add_conditional_edges(
"start",
route_next_action,
{
"process_input": "process_input_node",
"confirm_action": "confirm_action_node",
"notify_user": "notify_user_node",
"end": END,
"wait_for_user": "wait_for_user_node", # A node that just waits for new input
}
)
Each node's responsibility now included not just processing data, but also explicitly setting next_action based on its outcome. For example, a process_input_node might decide:
def process_input_node(state: AgentState):
# ... process input with LLM (Groq/Claude routing based on complexity)
if requires_confirmation:
state["next_action"] = "confirm_action"
elif task_is_done:
state["next_action"] = "notify_user"
else:
state["next_action"] = "wait_for_user" # Wait for more user input
return state
This pattern transformed my agents from fragile, implicit state machines into robust, explicit ones. The graph's flow became deterministic, driven by the next_action field. It also made debugging significantly easier: I could inspect the next_action in the checkpoint and immediately understand why the agent was transitioning (or not transitioning) to a particular node. This is crucial for production LangGraph stateful agents production checkpointing.
This approach also naturally supports multi-turn conversations where the agent needs to wait for user input. The wait_for_user action simply routes to a node that does nothing but return the state, effectively pausing the graph until new user_input arrives.
The Cost of Robustness
Frequently Asked Questions
Q: What's your strategy for routing between different LLMs (Groq, Claude) based on task complexity?
A: I use a small, fast LLM (e.g., Llama 3 8B on Groq) as a router. The router analyzes the user input and the current AgentState to determine if the task requires complex reasoning (routing to Claude 3 Opus) or can be handled by a cheaper, faster model (routing to Groq). This decision is part of the process_input_node logic.