Part 3: Scaling LangGraph: State Persistence, Checkpointing, and Parallelism

Long-running multi-agent workflows require durable state persistence and concurrency management. Here is how to configure SQLite and Redis checkpointers alongside Pregel super-step parallel branching in LangGraph.

HimanshuMay 03, 2024Updated September 08, 2026

An agent process throws midway through execution. Every message it accumulated and every intermediate reasoning step it produced lived in memory, so all of it goes with the process.

A graph without a checkpointer is not resumable. One exception discards every super-step that already succeeded, and a retry pays for all of them again.

A stateless workflow that restarts looks like it recovered. It started over. Anything you run in production needs checkpoint persistence, and once you have that, you can also run branches concurrently.

Chapter 0: Checkpointing mechanisms in graph runtimes

A checkpointer serializes the entire state of a graph to external storage (such as SQLite, Redis, or PostgreSQL) after each super-step. Stored data includes accumulated message lists, custom state variables, node execution markers, and routing metadata.

To resume execution after an interruption or process restart, the graph loads the state snapshot associated with a thread_id and continues execution from the last completed node.

Step 1: Local persistence with SqliteSaver

For single-instance deployments or local testing, SqliteSaver serializes state transitions directly into SQLite tables.

The thread_id is the primary lookup key for state snapshots:

python

import os
from typing import TypedDict, List
from langgraph.graph import StateGraph, START
from langgraph.checkpoint.sqlite import SqliteSaver
 
class GraphState(TypedDict):
    messages: List[str]
    turn: int
 
def chat_node(state: GraphState) -> dict:
    current_turn = state.get("turn", 0)
    new_message = f"Agent turn {current_turn}: Step complete."
    return {
        "messages": state.get("messages", []) + [new_message], 
        "turn": current_turn + 1
    }
 
memory = SqliteSaver.from_conn_string(":memory:")
 
workflow = StateGraph(GraphState)
workflow.add_node("chat", chat_node)
workflow.set_entry_point("chat")
workflow.add_edge("chat", "chat")
 
app = workflow.compile(checkpointer=memory)
 
thread_id = "session_001"
initial_state = {"messages": ["User: Run initial batch."], "turn": 0}
 
print(f"Executing thread '{thread_id}':")
for i in range(2):
    output = app.invoke(
        input=initial_state if i == 0 else {},
        config={"configurable": {"thread_id": thread_id}}
    )
    print(output['messages'][-1])
 
print("\nResuming from saved checkpoint:")
output_resumed = app.invoke(
    input={},
    config={"configurable": {"thread_id": thread_id}}
)
print("Resumed output:", output_resumed['messages'][-1])
print("Total history length:", len(output_resumed['messages']))

Invoking the compiled graph with an empty input dictionary and an existing thread_id automatically retrieves the latest checkpoint from SQLite and executes the next step.

Step 2: Distributed persistence with RedisSaver

In distributed environments where multiple worker nodes process tasks concurrently, single-file SQLite databases create file-lock contention.

RedisSaver moves state snapshots into a networked in-memory key-value store, so any available worker can pick up a suspended thread:

python

import os
import redis
from typing import TypedDict, List
from langgraph.graph import StateGraph
from langgraph.checkpoint.redis import RedisSaver
 
class GraphState(TypedDict):
    messages: List[str]
    turn: int
 
def chat_node(state: GraphState) -> dict:
    current_turn = state.get("turn", 0)
    new_message = f"Turn {current_turn} processed via Redis."
    return {
        "messages": state.get("messages", []) + [new_message], 
        "turn": current_turn + 1
    }
 
# Connect to Redis
redis_url = os.getenv("REDIS_URL", "redis://localhost:6379/0")
memory = RedisSaver.from_conn_info(host="localhost", port=6379, db=0)
 
workflow = StateGraph(GraphState)
workflow.add_node("chat", chat_node)
workflow.set_entry_point("chat")
workflow.add_edge("chat", "chat")
 
app_redis = workflow.compile(checkpointer=memory)
 
thread_id_redis = "distributed_thread_42"
initial_state = {"messages": ["Init distributed task."], "turn": 0}
 
output = app_redis.invoke(
    input=initial_state,
    config={"configurable": {"thread_id": thread_id_redis}}
)
print("First turn output:", output['messages'][-1])

Checkpointer	Storage engine	Concurrency model	Best fit
`SqliteSaver`	Local SQLite file / memory	Single-process only	Local development and tests
`RedisSaver`	Redis cluster	Multi-process networked workers	Low-latency distributed agents
`PostgresSaver`	PostgreSQL tables	Transactional ACID storage	High-durability enterprise workflows

Step 3: Branch parallelism via Pregel super-steps

LangGraph executes graphs using the Pregel computational model. Computations progress through synchronized super-steps:

In each super-step, all nodes whose inbound edge dependencies are met execute concurrently.
Nodes emit state updates and pass messages to downstream targets.
The super-step completes when all active nodes finish their execution cycle.

To execute tasks in parallel, define multiple outgoing edges from a single parent node without adding edges between the sibling workers:

python

from typing import List, Literal, TypedDict
from langgraph.graph import StateGraph, END
import time
 
class MultiAgentState(TypedDict):
    research_topics: List[str]
    report_sections: List[str]
    final_report: str
    current_step: Literal["plan", "research", "write", "finish"]
 
def plan_topics(state: MultiAgentState) -> dict:
    topics = ["AI Ethics", "Quantum Computing", "Neuroscience in AI"]
    print("Planner assigned topics:", topics)
    return {"research_topics": topics, "current_step": "research"}
 
def research_ai_ethics(state: MultiAgentState) -> dict:
    time.sleep(0.1)
    section = "AI Ethics: Evaluated algorithmic transparency and alignment."
    return {"report_sections": state.get("report_sections", []) + [section]}
 
def research_quantum_computing(state: MultiAgentState) -> dict:
    time.sleep(0.08)
    section = "Quantum Computing: Analyzed qubit coherence algorithms."
    return {"report_sections": state.get("report_sections", []) + [section]}
 
def research_neuroscience_ai(state: MultiAgentState) -> dict:
    time.sleep(0.12)
    section = "Neuroscience in AI: Modeled sparse activation graphs."
    return {"report_sections": state.get("report_sections", []) + [section]}
 
def compile_report(state: MultiAgentState) -> dict:
    sections = "\n".join(state.get("report_sections", []))
    report = f"Consolidated Findings:\n{sections}"
    return {"final_report": report, "current_step": "finish"}
 
workflow_parallel = StateGraph(MultiAgentState)
 
workflow_parallel.add_node("plan", plan_topics)
workflow_parallel.add_node("research_ethics", research_ai_ethics)
workflow_parallel.add_node("research_quantum", research_quantum_computing)
workflow_parallel.add_node("research_neuroscience", research_neuroscience_ai)
workflow_parallel.add_node("compile_report", compile_report)
 
workflow_parallel.set_entry_point("plan")
 
# Fan-out to independent research nodes
workflow_parallel.add_edge("plan", "research_ethics")
workflow_parallel.add_edge("plan", "research_quantum")
workflow_parallel.add_edge("plan", "research_neuroscience")
 
# Fan-in synchronization barrier: compile_report waits for all three nodes
workflow_parallel.add_edge("research_ethics", "compile_report")
workflow_parallel.add_edge("research_quantum", "compile_report")
workflow_parallel.add_edge("research_neuroscience", "compile_report")
 
workflow_parallel.add_edge("compile_report", END)
 
app_parallel = workflow_parallel.compile()

The runtime executes research_ethics, research_quantum, and research_neuroscience in parallel within the same super-step, waiting for all three to complete before triggering compile_report.

Step 4: Node-level async concurrency

Pregel coordinates high-level graph topology. However, I/O-bound tasks within individual nodes (such as querying multiple APIs or embedding text batches) should use native asynchronous Python concurrency:

python

import asyncio
import httpx
 
async def fetch_api_data(endpoints: List[str]) -> List[dict]:
    """Fetches multiple external endpoints concurrently using httpx."""
    async with httpx.AsyncClient() as client:
        tasks = [client.get(url, timeout=10.0) for url in endpoints]
        responses = await asyncio.gather(*tasks, return_exceptions=True)
        
        valid_results = []
        for r in responses:
            if isinstance(r, httpx.Response) and r.status_code == 200:
                valid_results.append(r.json())
        return valid_results

Using asyncio.gather inside node definitions prevents network round trips from blocking the overall Pregel event loop.

When this is the wrong choice

The whole run is one request and cheap to repeat. A checkpointer writes the full graph state to storage after every super-step. If re-running the workflow costs one model call, those writes buy you nothing and add a storage dependency to a process that had none.
The concurrency you need lives inside a single node. Splitting three API calls into three graph nodes gets you the same wall-clock time as asyncio.gather in one node, plus three checkpoint writes and a fan-in barrier. Fan out in the graph only when the branches are separately resumable.
The branches depend on each other. A super-step runs nodes whose inbound edges are satisfied. If the quantum research needs the ethics section first, the barrier serializes them anyway, and you have written a parallel-looking graph that executes sequentially.
You are on one process and staying there. SqliteSaver covers this. Redis and PostgreSQL exist for multiple workers contending on the same threads; adopting them earlier adds an operational service to keep alive for no concurrency you actually have.

Architecture rules for scaled graphs

State payload optimization: avoid storing raw binary files or extensive document corpora in checkpoint stores. Store pointers (such as S3 URIs or database primary keys) to prevent storage bloat.
Deterministic reducer policies: when multiple parallel nodes write to the same list field within a super-step, use explicit reducer functions to prevent race conditions.
Appropriate store selection: use SQLite for local unit testing, Redis for low-latency session caching, and PostgreSQL for long-term auditable checkpoints.