LangGraph Persistent State Across Runs

Free · Open source (MIT) · Works with LangChain, CrewAI, AutoGen · No signup

When your LangGraph workflow loses progress after a process restart, you're hitting the persistent state problem. Your carefully orchestrated agent nodes, their decisions, and accumulated context vanish the moment your process dies. This breaks long-running workflows, user conversations, and any stateful logic that spans multiple executions. Here's how to fix it with persistent checkpointing.

The Problem with Stateless LangGraph Workflows

LangGraph excels at building stateful, multi-step agent workflows. But by default, all that state lives in memory. When your Python process restarts—whether from a crash, deployment, or server reboot—your workflow's progress disappears.

Consider a multi-agent research workflow: Agent A gathers data, Agent B analyzes it, Agent C generates a report. If the process dies after Agent B finishes, you lose everything. The user gets a broken experience, and you waste compute re-running completed steps.

Traditional solutions like pickle files or local JSON storage break in distributed environments. Database setups add complexity. Redis requires infrastructure. You need persistent state that survives process restarts, works across machines, and requires zero configuration.

The core issue: LangGraph's default in-memory state storage isn't designed for production persistence. Your stateful graphs need a persistent backend that seamlessly integrates with LangGraph's checkpointer system.

The Fix: Persistent State with BotWire

Install BotWire Memory for zero-config persistent storage:

pip install botwire

Here's a working LangGraph workflow with persistent state:

from langgraph import StateGraph
from botwire import Memory
import json

# Initialize persistent memory
memory = Memory("my-workflow")

def save_checkpoint(thread_id: str, state: dict):
    """Save workflow state to persistent storage"""
    memory.set(f"checkpoint_{thread_id}", json.dumps(state))

def load_checkpoint(thread_id: str) -> dict:
    """Load workflow state from persistent storage"""
    data = memory.get(f"checkpoint_{thread_id}")
    return json.loads(data) if data else {}

# Your workflow now survives restarts
def my_workflow_node(state):
    # Process and return updated state
    return {"messages": state["messages"] + ["processed"]}

How Persistent State Works

The code above creates a persistent key-value store that survives process restarts. When your workflow hits a checkpoint, it serializes the current state and stores it remotely. On restart, it loads the last known state and continues execution.

BotWire handles the HTTP requests, data persistence, and error recovery behind the scenes. No signup, no API keys, no configuration files. Your data persists at https://botwire.dev with a free tier that covers most development and small production workloads.

def resume_workflow(thread_id: str):
    """Resume a workflow from its last checkpoint"""
    state = load_checkpoint(thread_id)
    
    if not state:
        # First run - initialize state
        state = {"messages": [], "step": 0}
    
    print(f"Resuming from step {state.get('step', 0)}")
    return state

def advanced_checkpoint_manager():
    """Handle multiple workflows with TTL and cleanup"""
    memory = Memory("workflows")
    
    # List all active workflows
    all_keys = memory.get("workflow_list") or "[]"
    active_workflows = json.loads(all_keys)
    
    # Clean up old workflows (manual TTL)
    for workflow_id in active_workflows:
        state = memory.get(f"checkpoint_{workflow_id}")
        if should_cleanup(state):  # Your cleanup logic
            memory.set(f"checkpoint_{workflow_id}", None)  # Delete

Cross-process coordination works seamlessly. Multiple workers can read the same workflow state, and the last writer wins. For complex scenarios, implement your own locking mechanism using BotWire's atomic operations.

Integration with LangGraph Checkpointers

LangGraph provides a checkpointer interface for persistent state. Here's how to build a BotWire-backed checkpointer:

from langgraph.checkpoint import BaseCheckpointSaver
from botwire import Memory
import json
from typing import Optional, Dict, Any

class BotWireCheckpointSaver(BaseCheckpointSaver):
    def __init__(self, namespace: str = "langgraph"):
        self.memory = Memory(namespace)
    
    def get_tuple(self, thread_id: str) -> Optional[Dict[Any, Any]]:
        """Load checkpoint for a thread"""
        data = self.memory.get(f"thread_{thread_id}")
        return json.loads(data) if data else None
    
    def put_tuple(self, thread_id: str, checkpoint: Dict[Any, Any]) -> None:
        """Save checkpoint for a thread"""
        self.memory.set(f"thread_{thread_id}", json.dumps(checkpoint))
    
    def list_tuples(self, thread_id: str, limit: int = 10):
        """List recent checkpoints - simplified version"""
        checkpoint = self.get_tuple(thread_id)
        return [checkpoint] if checkpoint else []

# Use with your LangGraph workflow
checkpointer = BotWireCheckpointSaver()
graph = StateGraph(YourState).compile(checkpointer=checkpointer)

When NOT to Use BotWire

BotWire isn't the right choice for every persistence need:

• Vector search or semantic similarity - Use Pinecone, Weaviate, or ChromaDB for embeddings and semantic search • High-throughput applications - Free tier limits to 1000 writes/day; Redis or dedicated databases handle higher loads better • Sub-millisecond latency requirements - Network round trips add ~50-200ms; use in-memory caching for ultra-low latency

FAQ

Q: Why not just use Redis for LangGraph state? A: Redis requires setup, infrastructure, and connection management. BotWire works immediately with zero configuration, perfect for development and small production workloads.

Q: Is this actually free? A: Yes, the free tier is permanent: 1000 writes/day per namespace, 50MB storage, unlimited reads. No credit card required.

Q: What about data privacy? A: BotWire is open source (MIT license) and self-hostable. For sensitive data, run your own instance using the single FastAPI + SQLite service.

Get Started with Persistent LangGraph State

Stop losing workflow progress on every restart. Install BotWire and add persistent state to your LangGraph workflows in minutes: pip install botwire. Check the full documentation and self-hosting options at https://botwire.dev.

Install in one command:

pip install botwire

Start free at botwire.dev

Related guides