Agent skill

langgraph-persistence

Persistence and human-in-the-loop patterns for LangGraph. Use when implementing checkpointing, saving/restoring graph state, adding human approval workflows, managing conversation memory across sessions, enabling time-travel debugging, or building fault-tolerant agents that resume from failures.

View SKILL.md on GitHub Repository
Stars 163
Forks 31

Install this agent skill to your Project

npx add-skill https://github.com/majiayu000/claude-skill-registry/tree/main/skills/development/langgraph-persistence

SKILL.md

LangGraph Persistence

Persistence enables memory, human-in-the-loop, time-travel, and fault-tolerance.

Checkpointers

Save graph state at every super-step. Available implementations:

Checkpointer Package Use Case
InMemorySaver langgraph-checkpoint Development/testing
SqliteSaver langgraph-checkpoint-sqlite Local workflows
PostgresSaver langgraph-checkpoint-postgres Production

Basic Setup

python
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import StateGraph

checkpointer = InMemorySaver()
graph = StateGraph(State)
# ... add nodes/edges ...
app = graph.compile(checkpointer=checkpointer)

Threads

Unique identifier for each conversation/execution. Required for persistence.

python
config = {"configurable": {"thread_id": "user-123-session-1"}}

# First invocation
result = app.invoke({"messages": [msg]}, config)

# Continue same thread (retains history)
result = app.invoke({"messages": [new_msg]}, config)

PostgreSQL (Production)

python
from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver

async with AsyncPostgresSaver.from_conn_string(DATABASE_URL) as checkpointer:
    await checkpointer.setup()  # Create tables
    app = graph.compile(checkpointer=checkpointer)

Human-in-the-Loop

Static Interrupts

Pause at predefined points:

python
# Pause BEFORE node executes
app = graph.compile(
    checkpointer=checkpointer,
    interrupt_before=["sensitive_action"]
)

# Pause AFTER node executes
app = graph.compile(
    checkpointer=checkpointer,
    interrupt_after=["review_step"]
)

Execution and resume:

python
config = {"configurable": {"thread_id": "1"}}

# Run until interrupt
for event in app.stream(inputs, config):
    print(event)
# Graph pauses at interrupt_before node

# ... human reviews state ...

# Resume from checkpoint (pass None as input)
for event in app.stream(None, config):
    print(event)

Dynamic Interrupts

Pause based on runtime conditions using interrupt():

python
from langgraph.types import interrupt, Command

def sensitive_node(state: State) -> dict:
    if state["requires_approval"]:
        # Pause and request human input
        human_response = interrupt({
            "question": "Approve this action?",
            "proposed_action": state["action"]
        })
        if human_response == "reject":
            return {"status": "cancelled"}
    return {"status": "approved"}

Resume with Command:

python
# Resume with human response
app.invoke(Command(resume="approve"), config)

Editing State

Modify state before resuming:

python
# Get current state
state = app.get_state(config)

# Update state
app.update_state(
    config,
    {"approved": True, "reviewer": "admin"},
    as_node="approval_node"  # Credit update to this node
)

# Resume
app.invoke(None, config)

Common Patterns

Approval Pattern

python
from langgraph.types import interrupt

def approval_node(state: State) -> dict:
    response = interrupt({
        "type": "approval_request",
        "details": state["pending_action"]
    })
    if response["approved"]:
        return {"status": "approved"}
    return {"status": "rejected", "reason": response.get("reason")}

Tool Call Review

python
def review_tools(state: State) -> dict:
    last_msg = state["messages"][-1]
    if last_msg.tool_calls:
        decision = interrupt({
            "tool_calls": last_msg.tool_calls,
            "question": "Execute these tool calls?"
        })
        if not decision["proceed"]:
            # Modify or reject tool calls
            return {"messages": [modified_response]}
    return {}

Multi-Turn Input Collection

python
def collect_info(state: State) -> Command:
    missing = get_missing_fields(state)
    if missing:
        user_input = interrupt({
            "missing_fields": missing,
            "prompt": f"Please provide: {missing}"
        })
        return Command(update={"user_data": user_input})
    return Command(goto="process")

Long-Term Memory (Store)

For cross-thread persistent storage:

python
from langgraph.store.memory import InMemoryStore

store = InMemoryStore()
app = graph.compile(
    checkpointer=checkpointer,
    store=store
)

# Access in node via runtime
def node_with_memory(state: State, *, store) -> dict:
    # Store/retrieve user preferences across threads
    user_prefs = store.get(("users", state["user_id"]))
    return {"preferences": user_prefs}

State Inspection

python
# Get current state snapshot
snapshot = app.get_state(config)
print(snapshot.values)  # Current state
print(snapshot.next)    # Next nodes to execute

# Get state history
for state in app.get_state_history(config):
    print(state.config, state.values)

# Time travel: resume from specific checkpoint
old_config = {"configurable": {
    "thread_id": "1",
    "checkpoint_id": "abc123"
}}
app.invoke(None, old_config)

Key Points

  • Always use thread_id in config for persistence
  • interrupt_before/interrupt_after for static pauses
  • interrupt() function for dynamic, conditional pauses
  • Resume with None input or Command(resume=...)
  • update_state() to edit state before resuming
  • PostgresSaver for production deployments

Didn't find tool you were looking for?

Be as detailed as possible for better results