How to Rollback State in LangGraph

The Danger of Native Checkpointers

LangGraph is an incredible framework for building multi-agent architectures. However, its native memory systems (like MemorySaver, SqliteSaver, and PostgresSaver) have a fatal flaw for production applications: they assume all agent generations are inherently safe and valid.

If an LLM hallucinates and outputs an invalid JSON payload, LangGraph will happily save that corrupted data to your database as the new "current state". Once corrupted state is committed to the database, your agent workflow is effectively dead. The next time an agent tries to read the state, the entire graph crashes with parsing errors or logical contradictions.

The Solution: Transactional Memory

To safely deploy AI to production, you must treat agent memory exactly like a financial database: using Transactions.

A transaction guarantees ACID properties (Atomicity, Consistency, Isolation, Durability). In the context of AI agents, Consistency is the most critical. If an agent attempts an action that violates your business rules, the transaction must fail, and the state must roll back to exactly where it was before the agent made the mistake.

Why typical error catching fails

You might think you can just catch the error in a node:

def my_node(state):
    try:
        new_value = llm.invoke(...)
        # Validate new_value
        if not is_valid(new_value):
            raise Exception("Bad LLM output")
        return {"data": new_value}
    except Exception as e:
        return {"error": str(e)} # State is still updated with the error!

The problem here is that the state still mutates. You now have an error key floating around your graph, and subsequent nodes have to deal with it. You haven't rolled back; you've just moved the problem down the line.

Implementing Real Rollbacks in 2 Lines of Code

You don't need to write a massive custom checkpointer to achieve true state rollbacks. You can use StateGuard, a drop-in replacement for LangGraph's native memory checkpointers.

1. Define your Invariants

First, define what constitutes "valid" state in your application.

from stateguard import StateGuardCheckpointer

# Define a strict invariant rule
def no_negative_balance(state):
    if state.get("balance", 0) < 0:
        raise ValueError("Agents cannot overdraft accounts!")
    return True

2. Swap the Checkpointer

Instead of passing MemorySaver() to your graph compiler, pass the StateGuardCheckpointer.

# 1. Initialize StateGuard with your invariants
checkpointer = StateGuardCheckpointer(
    invariants=[no_negative_balance],
    # You can also pass a persistent backend like Postgres
    backend="sqlite:///memory.db" 
)

# 2. Compile your existing LangGraph workflow
app = workflow.compile(checkpointer=checkpointer)

Now, if an agent hallucinates a negative balance in any node, StateGuard intercepts the database write. The write is blocked, a RollbackEvent is triggered, and the thread state remains exactly as it was at the start of the step. The corrupt data never touches your database.

Advanced: Multi-Node Rollbacks (Savepoints)

Sometimes, an error in Node C means you need to rollback all the way to the state before Node A started. StateGuard supports explicit savepoints, similar to PostgreSQL.

def node_a(state):
    # Create a manual savepoint before starting a risky multi-step operation
    state["__stateguard_savepoint"] = "pre_risky_operation"
    return {"status": "started"}

def node_c(state):
    if some_validation_fails():
        # Instruct StateGuard to rollback to the specific savepoint
        return {"__stateguard_rollback_to": "pre_risky_operation"}

Conclusion

By enforcing transactional boundaries at the checkpointer level, you completely decouple your graph logic from your safety logic. Your agents can run fast and loose, and StateGuard acts as the impenetrable firewall that protects your database.